package br

import (
	"context"
	"fmt"
	"path"
	"sort"
	"strings"
	"time"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/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/clustergateway"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"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"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/api/resource"
)

func resolveTiDBClusterKeyspaceForBackup(sharedPool *entity.SharedPool, clusterID string) string {
	return clusterID
}

const (
	DefaultStorageClass                 = "cloud-ssd"
	DefaultTiBRCPU                      = "2"
	DefaultTiBRMemory                   = "8Gi"
	DefaultTiBRStorageSize              = "500Gi" // it's safe for big cluster restore
	DefaultTibrgcT3StorageSize          = "250Gi"
	DefaultTiKVUsagePercentageThreshold = 70
)

const (
	DefaultMinT2DataAgeDays        uint32 = 7
	DefaultMinT3DataAgeDays        uint32 = 3
	DefaultMaxT1MetaLifetimeDays   uint32 = 7
	DefaultMaxT1WalLifetimeDays    uint32 = 7
	DefaultMaxT1PackedLifetimeDays uint32 = 4
	DefaultMaxT2SstLifetimeDays    uint32 = 30
	DefaultMaxT3DataLifetimeDays   uint32 = 90
)

var (
	DefaultClassicRestoreImageTag = "v8.5.4-nextgen.202510.10"
	DefaultClassicRestoreOptions  = map[string]string{"with-sys-table": "true"}
)

const (
	KernelPackBackupTaskId    = "101"
	KernelRestorePackedTaskId = "202"
	KernelCopyBackupTaskId    = "303"
)

func AddActionForLWRestore(ctx context.Context, rp repo.LightweightRestoreRepo, restoreID, action string) error {
	added, err := rp.AddExecutedAction(ctx, restoreID, action)
	if err != nil {
		return fmt.Errorf("failed to add action to restore: %w", err)
	}
	if added {
		tilog.Info(ctx, "add action to restore success", zap.String("action", action), zap.String("restoreID", restoreID))
	}
	return nil
}

func AddActionForClassicRestore(ctx context.Context, rp repo.ClassicRestoreRepo, restoreID, action string) error {
	added, err := rp.AddExecutedAction(ctx, restoreID, action)
	if err != nil {
		return fmt.Errorf("failed to add action to restore: %w", err)
	}
	if added {
		tilog.Info(ctx, "add action to restore success", zap.String("action", action), zap.String("restoreID", restoreID))
	}
	return nil
}

func AddActionForManualBackup(ctx context.Context, rp repo.ManualBackupRepo, manualBackup *entity.ManualBackup, action string) error {
	added, err := rp.AddExecutedAction(ctx, manualBackup.ClusterID, manualBackup.BackupID, action)
	if err != nil {
		return fmt.Errorf("failed to add action to manual backup: %w", err)
	}
	if added {
		tilog.Info(ctx, "add action to manual backup success", zap.String("action", action), zap.String("clusterID", manualBackup.ClusterID), zap.String("backupID", manualBackup.BackupID))
	}
	return nil
}

func kernelRestoreFailure(code, message string) *entity.RestoreFailure {
	return &entity.RestoreFailure{
		Source:  entity.RestoreFailureSourceKernel,
		Code:    code,
		Message: message,
	}
}

func controlPlaneRestoreFailure(code, message string) *entity.RestoreFailure {
	return &entity.RestoreFailure{
		Source:  entity.RestoreFailureSourceControlPlane,
		Code:    code,
		Message: message,
	}
}

func GetRestoreID(args actions.ActionArgs) (restoreID string, existed bool) {
	restoreIDArg, ok := args[workflow.RestoreIdKey]
	if !ok {
		return "", false
	}
	return restoreIDArg.String(), true
}

func GetManualBackupID(args actions.ActionArgs) (backupID string, existed bool) {
	backupIDArg, ok := args[workflow.ManualBackupIdKey]
	if !ok {
		return "", false
	}
	return backupIDArg.String(), true
}

func GetManualRestoreID(args actions.ActionArgs) (restoreID string, existed bool) {
	restoreIDArg, ok := args[workflow.ManualRestoreIdKey]
	if !ok {
		return "", false
	}
	return restoreIDArg.String(), true
}

func GetTiDBClusterID(args actions.ActionArgs) (tidbClusterID string, existed bool) {
	tidbClusterIDArg, ok := args[workflow.TiDBClusterIdKey]
	if !ok {
		return "", false
	}
	return tidbClusterIDArg.String(), true
}

func AddActionForManualRestore(ctx context.Context, rp repo.ManualBackupRestoreRepo, restoreID, action string) error {
	added, err := rp.AddExecutedAction(ctx, restoreID, action)
	if err != nil {
		return fmt.Errorf("failed to add action to manual restore: %w", err)
	}
	if added {
		tilog.Info(ctx, "add action to manual restore success", zap.String("action", action), zap.String("restoreID", restoreID))
	}
	return nil
}

func GetDataplaneID(args actions.ActionArgs) (dataplaneID string, existed bool) {
	dataplaneIDArg, ok := args[workflow.DataplaneIdKey]
	if !ok {
		return "", false
	}
	return dataplaneIDArg.String(), true
}

type BucketPermissionContext struct {
	DataplaneID          string
	BucketID             string
	PermittedDataplaneID string
}

func getBucketPermissionContext(ctx context.Context,
	restoreID string,
	lwRepo repo.LightweightRestoreRepo,
	spRepo repo.SharedPoolRepo,
	osRepo repo.SharedPoolObjectStoreRepo,
) (*BucketPermissionContext, error) {
	restore, err := lwRepo.Load(ctx, restoreID, false)
	if err != nil {
		return nil, fmt.Errorf("failed to load lw restore: %w", err)
	}
	targetSharedPool, err := spRepo.Load(ctx, restore.TargetSharedPoolID, false)
	if err != nil {
		return nil, fmt.Errorf("failed to load targetSharedPool: %w", err)
	}
	sourceSharedPool, err := spRepo.Load(ctx, restore.SourceSharedPoolID, true)
	if err != nil {
		return nil, fmt.Errorf("failed to load sourceSharedPool: %w", err)
	}
	sourceSharedPoolObjectStore, err := osRepo.Load(ctx, restore.SourceSharedPoolID, schema.SharedPoolObjectStoreTypeData, true)
	if err != nil {
		return nil, fmt.Errorf("failed to load sourceSharedPoolObjectStore: %w", err)
	}
	return &BucketPermissionContext{
		DataplaneID:          sourceSharedPool.GetDataplaneName(),
		BucketID:             sourceSharedPoolObjectStore.StoreBucket,
		PermittedDataplaneID: targetSharedPool.GetDataplaneName(),
	}, nil
}

// includeDeletedTargetPool should be true only for the revoke (cleanup) path:
// revoking the grant of a deleted target dataplane is still meaningful, while
// granting access to a deleted pool never is.
func getBucketPermissionContextForManualRestore(ctx context.Context,
	restoreID string,
	includeDeletedTargetPool bool,
	manualRestoreRepo repo.ManualBackupRestoreRepo,
	spRepo repo.SharedPoolRepo,
	osRepo repo.SharedPoolObjectStoreRepo,
) (*BucketPermissionContext, error) {
	restore, err := manualRestoreRepo.Load(ctx, restoreID, false)
	if err != nil {
		return nil, fmt.Errorf("failed to load manual backup restore: %w", err)
	}
	targetSharedPool, err := spRepo.Load(ctx, restore.TargetSharedPoolID, includeDeletedTargetPool)
	if err != nil {
		return nil, fmt.Errorf("failed to load targetSharedPool: %w", err)
	}
	sourceSharedPool, err := spRepo.Load(ctx, restore.SourceSharedPoolID, true)
	if err != nil {
		return nil, fmt.Errorf("failed to load sourceSharedPool: %w", err)
	}
	sourceSharedPoolObjectStore, err := osRepo.Load(ctx, restore.SourceSharedPoolID, schema.SharedPoolObjectStoreTypeData, true)
	if err != nil {
		return nil, fmt.Errorf("failed to load sourceSharedPoolObjectStore: %w", err)
	}
	return &BucketPermissionContext{
		DataplaneID:          sourceSharedPool.GetDataplaneName(),
		BucketID:             sourceSharedPoolObjectStore.StoreBucket,
		PermittedDataplaneID: targetSharedPool.GetDataplaneName(),
	}, nil
}

func ParseToRFC3339(timestr string) (string, error) {
	if _, err := time.Parse(time.RFC3339, timestr); err == nil {
		return timestr, nil
	}

	inputLayout := "2006-01-02 15:04:05.000"
	t, err := time.Parse(inputLayout, timestr)
	if err != nil {
		return "", fmt.Errorf("failed to parse time: %w", err)
	}
	output := t.Format(time.RFC3339)
	return output, nil
}

func applyAzureRestorePackedBackupSource(
	ctx context.Context,
	targetSharedPool *entity.SharedPool,
	sourceSharedPoolID string,
	spRepo repo.SharedPoolRepo,
	osRepo repo.SharedPoolObjectStoreRepo,
	bucketSvc dplanemgrv1.BucketServiceClient,
	req *clustergateway.CreateRestorePackedBackupTaskRequest,
) error {
	// Source shared pools are loaded with deleted=true in x-restore paths. After soft deletion,
	// the loaded source shared pool may no longer carry Provider.Type, so use the target shared
	// pool to decide whether Azure-specific endpoint handling is needed.
	if targetSharedPool == nil || targetSharedPool.GetProvider().GetType() != apiv1.ProviderType_AZURE {
		return nil
	}

	sourceBucket, _, err := GetBucketInfoOfSharedPool(ctx, sourceSharedPoolID, spRepo, osRepo, bucketSvc)
	if err != nil {
		return fmt.Errorf("failed to get source bucket info of shared pool %s: %w", sourceSharedPoolID, err)
	}

	parts := strings.SplitN(sourceBucket, "/", 2)
	if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
		return fmt.Errorf("failed to parse azure bucket format %s of source shared pool %s: %w",
			sourceBucket, sourceSharedPoolID,
			fmt.Errorf("invalid azure bucket format %s, expected <account>/<container>", sourceBucket))
	}
	req.ExoticBackupEndpoint = fmt.Sprintf("https://%s.blob.core.windows.net", parts[0])
	return nil
}

func resolveSourceBucketForTargetSharedPool(targetSharedPool *entity.SharedPool, bucket string) (useS3Bucket, useEndpoint string, err error) {
	if targetSharedPool == nil || targetSharedPool.GetProvider().GetType() != apiv1.ProviderType_AZURE {
		return bucket, "", nil
	}

	parts := strings.Split(bucket, "/")
	if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
		return "", "", fmt.Errorf("invalid azure bucket format %s, expected <account>/<container>", bucket)
	}

	return parts[1], fmt.Sprintf("https://%s.blob.core.windows.net", parts[0]), nil
}

func GetImage(targetSharedPool *entity.SharedPool, imageConfig *config.ImageConfig) (string, error) {
	for _, comp := range targetSharedPool.GetComponentGroups() {
		if comp.GetComponentType() == apiv1.ComponentType_TIKV {
			return fmt.Sprintf("%s/%s/%s:%s",
				imageConfig.Registry,
				imageConfig.RepoPrefix,
				path.Base(comp.GetConfiguration().GetImageRepo()),
				comp.GetConfiguration().GetImageTag()), nil
		}
	}
	return "", fmt.Errorf("tikv component not found in target shared pool %s", targetSharedPool.GetId())
}

func GetStorageClass(tibrConfig *config.TiBRConfig, providerType apiv1.ProviderType) string {
	if tibrConfig != nil {
		if tibrConfig.StorageClass != "" {
			return tibrConfig.StorageClass
		}
	}
	return DefaultStorageClass
}

func GetTiKVCpuMemorySpec(ctx context.Context, sharedPool *entity.SharedPool, dataplaneManager dataplane.Manager) (string, string, error) {
	remoteCli, err := dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return "", "", fmt.Errorf("failed to get remote client for shared pool: %w", err)
	}
	tikgs, err := tioperator.ListTiKVGroups(ctx, remoteCli, sharedPool.GetId())
	if err != nil {
		return "", "", fmt.Errorf("failed to list tikv groups: %w", err)
	}
	if len(tikgs) == 0 {
		return "", "", fmt.Errorf("no tikv group found for shared pool: %s", sharedPool.GetId())
	}
	tikvGroup := tikgs[0]
	if tikvGroup.GetNodeQuota() == nil || tikvGroup.GetNodeQuota().GetNodeSpec() == nil {
		return "", "", fmt.Errorf("tikv group %s has no node spec", tikvGroup.GetId())
	}
	cpu := fmt.Sprintf("%d", tikvGroup.GetNodeQuota().GetNodeSpec().GetCpu())
	memory := fmt.Sprintf("%dGi", tikvGroup.GetNodeQuota().GetNodeSpec().GetMemoryGi())
	return cpu, memory, nil
}

func GetAvailableTiBRGCCpuMemory(ctx context.Context, sharedPool *entity.SharedPool, dataplaneManager dataplane.Manager) (uint64, string, error) {
	cpu, memory, err := GetTiKVCpuMemorySpec(ctx, sharedPool, dataplaneManager)
	if err != nil {
		return 0, "", fmt.Errorf("failed to get tikv cpu memory spec: %w", err)
	}

	milliCPU, err := ParseCPU(cpu)
	if err != nil {
		return 0, "", fmt.Errorf("failed to parse cpu %s: %w", cpu, err)
	}
	memoryInMiB, err := ParseToMiB(memory)
	if err != nil {
		return 0, "", fmt.Errorf("failed to parse memory %s: %w", memory, err)
	}
	availableMemory := tioperator.ComputeAvailableMemory(memoryInMiB)
	availableCPU := ComputeAvailableCPU(uint64(milliCPU) / 1000)

	return availableCPU, fmt.Sprintf("%dMi", availableMemory), nil
}

func GetArchiveBackupStorageSize(tibrConfig *config.TiBRConfig) string {
	if tibrConfig != nil {
		if tibrConfig.ArchiveBackupStorageSize != "" {
			return tibrConfig.ArchiveBackupStorageSize
		}
	}
	return DefaultTibrgcT3StorageSize
}

// mem example: "8Gi", "8192Mi"
func ParseToMiB(mem string) (uint64, error) {
	q, err := resource.ParseQuantity(mem)
	if err != nil {
		return 0, err
	}
	return uint64(q.Value() / (1024 * 1024)), nil
}

// core example: "500m", "1", "0.25"
func ParseCPU(cpuStr string) (milliCPU int64, err error) {
	q, err := resource.ParseQuantity(cpuStr)
	if err != nil {
		return 0, err
	}
	return q.MilliValue(), nil
}

func ComputeAvailableCPU(coreNum uint64) uint64 {
	if coreNum <= 1 {
		return 1
	}
	if coreNum%2 == 0 {
		return coreNum - 1
	}
	return coreNum
}

// Notice: include deleted shared pool
func GetBucketInfoOfSharedPool(
	ctx context.Context,
	sharedPoolId string,
	spRepo repo.SharedPoolRepo,
	osRepo repo.SharedPoolObjectStoreRepo,
	bucketSvc dplanemgrv1.BucketServiceClient,
) (bucketName, dfsPrefix string, err error) {
	sharedPool, err := spRepo.Load(ctx, sharedPoolId, true)
	if err != nil {
		tilog.Error(ctx, "failed to load shared pool", zap.Error(err), zap.String("sharedPoolId", sharedPoolId))
		return "", "", fmt.Errorf("failed to load shared pool: %w", err)
	}

	sharedPoolObjectStore, err := osRepo.Load(ctx, sharedPoolId, schema.SharedPoolObjectStoreTypeData, true)
	if err != nil {
		tilog.Error(ctx, "failed to load data sharedPoolObjectStore", zap.Error(err), zap.String("sharedPoolId", sharedPoolId))
		return "", "", fmt.Errorf("failed to load sharedPoolObjectStore: %w", err)
	}

	bucketId := sharedPoolObjectStore.StoreBucket
	if bucketId == "" {
		bucketId = entity.GetDataBucketId(sharedPoolId, sharedPoolObjectStore.StoreRegion)
		tilog.Warn(ctx, "bucket id is empty in sharedPoolObjectStore, use generated bucket id", zap.String("sharedPoolId", sharedPoolId), zap.String("bucketId", bucketId))
	}

	bucket, err := bucketSvc.GetBucket(ctx, &dplanemgrv1.GetBucketRequest{
		DataplaneId: sharedPool.GetDataplaneName(),
		BucketId:    bucketId,
	})
	if err != nil {
		tilog.Error(ctx, "failed to get bucket from dataplane manager", zap.Error(err), zap.String("sharedPoolId", sharedPoolId), zap.String("bucketId", bucketId))
		return "", "", fmt.Errorf("failed to get bucket: %w", err)
	}

	sharedPool.ObjectStorage = &entity.ObjectStorage{
		Prefix: sharedPoolObjectStore.StorePrefix,
		Bucket: bucket.GetName(),
		Region: bucket.GetRegion(),
	}

	return sharedPool.GetObjectStorage().GetBucket(), sharedPool.GetDfsPrefix(), nil
}

// required by o11y in metrics scrape for sharedpool level restore, which will be value of label app.kubernetes.io/instance in pod
func getComponentGroupName4TiBR(sharedPool *entity.SharedPool) string {
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() == apiv1.ComponentType_TIKV {
			return entity.GetComponentGroupName(comp)
		}
	}
	return ""
}

func getTenantTagValueOfDataplane(ctx context.Context, dpServiceCli dplanemgrv1.DataPlaneServiceClient, dataplaneId string) (string, error) {
	dp, err := dpServiceCli.GetDataPlane(ctx, &dplanemgrv1.GetDataPlaneRequest{
		DataPlaneId: dataplaneId,
	})
	if err != nil {
		return "", fmt.Errorf("failed to get dataplane: %w", err)
	}
	tags := dp.GetTags()
	if tags == nil {
		return "", fmt.Errorf("tags is nil for dataplane: %s", dataplaneId)
	}
	tenant, ok := tags[schemav1.LabelUserTenantKey]
	if !ok {
		return "", fmt.Errorf("tenant tag is not found for dataplane: %s", dataplaneId)
	}
	return tenant, nil
}

func genOptionsStr(options map[string]string) string {
	if len(options) == 0 {
		return "[]"
	}
	optionsArr := []string{}
	for key, value := range options {
		optionsArr = append(optionsArr, fmt.Sprintf("--%s=%s", key, value))
	}
	sort.Strings(optionsArr)
	//nolint: gocritic // not works if use %q
	return fmt.Sprintf(`["%s"]`, strings.Join(optionsArr, `", "`))
}

func classicRestoreSecretName(restoreID string) string {
	return fmt.Sprintf("cl-restore-%s-secret", restoreID)
}

// checkManualRestoreNotTerminated returns a TerminalError when the manual
// restore has reached a terminal state (cancelled / failed / soft-deleted).
// Actions of the create-restore DAGs call it at the beginning of Start and
// CheckFinish so they stop retrying once the restore can never make progress
// again (e.g. the restore was cancelled, or a previous check marked it failed).
func checkManualRestoreNotTerminated(ctx context.Context, rp repo.ManualBackupRestoreRepo, restoreID string) error {
	restore, err := rp.Load(ctx, restoreID, true)
	if err != nil {
		return fmt.Errorf("failed to load manual backup restore: %w", err)
	}
	if restore.DeletedAt.Valid {
		return actions.NewTerminalError(fmt.Errorf("manual restore %s is deleted, stop retrying", restoreID))
	}
	switch restore.Status {
	case schema.ManualBackupRestoreStatusCancelled, schema.ManualBackupRestoreStatusFailed:
		return actions.NewTerminalError(fmt.Errorf("manual restore %s is in terminal status %s, stop retrying", restoreID, restore.Status))
	}
	return nil
}