package task

import (
	"context"
	"fmt"
	"time"

	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/samber/lo"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"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"
	metricsctrl "github.com/tidbcloud/infra-provider/pkg/metrics/controller"
)

const (
	// physicalFileCacheMinAvailableMemoryGiB is the minimum total instance memory
	// (GiB) required to enable PhysicalFileCacheCapacity.
	physicalFileCacheMinAvailableMemoryGiB int64 = 230
	// physicalFileCacheReservedMemoryGiB is the memory headroom (GiB) reserved for
	// non-cache usage when calculating PhysicalFileCacheCapacity.
	physicalFileCacheReservedMemoryGiB int64 = 106
)

// CompactTask handles CompactBackup-specific operations
type CompactTask struct {
	*baseTask
}

// NewCompact creates a new CompactTask
func NewCompact(c *Context) deploy.Task {
	return &CompactTask{baseTask: &baseTask{Context: c}}
}

// Name returns the task name
func (t *CompactTask) Name() string {
	return "Compact"
}

// Category returns the task category
func (t *CompactTask) Category() deploy.TaskCategory {
	return deploy.BRTask
}

// Create implements the CompactBackup creation logic
func (t *CompactTask) Create(ctx context.Context) deploy.Result {
	start := time.Now()
	defer func() {
		metricsctrl.ReconcileDuration.WithLabelValues("compact_backup_create").Observe(time.Since(start).Seconds())
	}()

	t.logger.Info("Creating TiDB Operator CompactBackup", "name", t.compactBackup.Name)

	// Validate storage configuration
	if t.compactBackup.Spec.Storage == nil || *t.compactBackup.Spec.Storage == "" {
		t.logger.Error(nil, "compact backup storage cannot be empty")
		return deploy.Panic(fmt.Errorf("compact backup storage cannot be empty"))
	}

	// make the compact backup pod use the desired nodepool
	affinity := t.cloud.GetPodAffinity()
	instanceType := defaultInstanceType
	if specInstanceType := lo.FromPtr(t.compactBackup.Spec.InstanceType); specInstanceType != "" {
		instanceType = specInstanceType
	}

	compactBackup := &opv1alpha1.CompactBackup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.getCompactBackupCRName(),
			Namespace: t.clusterNamer.GetTiDBNamespace(),
			Labels:    t.compactBackup.Labels,
		},
		Spec: opv1alpha1.CompactSpec{
			StorageProvider: opv1alpha1.StorageProvider{},
			BR: &opv1alpha1.BRConfig{
				Cluster:          t.clusterNamer.GetTiDBName(),
				ClusterNamespace: t.clusterNamer.GetTiDBNamespace(),
				StatusAddr:       "0.0.0.0:8089",
			},
			Affinity:    affinity,
			Tolerations: utils.Toleration(cloudv1.ComponentCompactBackup),
			UseKMS:      false,
			Name:        t.getCompactName(),
		},
	}

	if capacity := t.getPhysicalFileCacheCapacity(ctx, instanceType); capacity != "" {
		compactBackup.Spec.PhysicalFileCacheCapacity = capacity
		t.logger.Info("configured physical file cache capacity for compact backup",
			"instanceType", instanceType,
			"physicalFileCacheCapacity", capacity)
	}

	// Set compact-specific fields
	if t.compactBackup.Spec.StartTs != nil {
		compactBackup.Spec.StartTs = *t.compactBackup.Spec.StartTs
	}
	if t.compactBackup.Spec.EndTs != nil {
		compactBackup.Spec.EndTs = *t.compactBackup.Spec.EndTs
	}
	if t.compactBackup.Spec.Concurrency != nil {
		compactBackup.Spec.Concurrency = int(*t.compactBackup.Spec.Concurrency)
	} else {
		compactBackup.Spec.Concurrency = 4 // default concurrency
	}

	if shard := lo.FromPtr(t.compactBackup.Spec.ShardCount); shard > 0 {
		compactBackup.Spec.Mode = opv1alpha1.CompactModeSharded
		compactBackup.Spec.ShardCount = &shard
	}

	// override image
	if t.compactBackup.Spec.Configuration != nil {
		config := t.compactBackup.Spec.Configuration

		if config.BaseImage != "" && config.Tag != "" {
			baseImage := gitopssdk.OverrideBaseImage(t.compactBackup, config.BaseImage)
			compactBackup.Spec.ToolImage = fmt.Sprintf("%s:%s", baseImage, config.Tag)
		}

		if config.TiKVImage != "" {
			tikvImage := config.TiKVImage
			if config.TiKVVersion != "" {
				tikvImage = fmt.Sprintf("%s:%s", config.TiKVImage, config.TiKVVersion)
			}
			compactBackup.Spec.TiKVImage = tikvImage
		}

		if config.Config != "" {
			if err := configmanager.MergeBRConfig(compactBackup.Spec.BR, []byte(config.Config)); err != nil {
				t.logger.Error(err, "failed to merge BR config")
				return deploy.Panic(err)
			}
		}
	}

	// Set resource requirements if specified
	if t.compactBackup.Spec.ResourceRequirements != nil {
		compactBackup.Spec.ResourceRequirements = *t.compactBackup.Spec.ResourceRequirements
		t.logger.Info("Applied resource requirements to compact backup",
			"requests", compactBackup.Spec.ResourceRequirements.Requests,
			"limits", compactBackup.Spec.ResourceRequirements.Limits)
	}

	// cloud vendor specific overlay - use the new CompactBackup specific method
	if err := t.cloud.OverlayCompact(compactBackup); err != nil {
		t.logger.Error(err, "failed to apply cloud provider overlay")
		return deploy.Panic(err)
	}

	err := t.applyOpCompactBackup(ctx, compactBackup)
	if err != nil {
		t.logger.Error(err, "failed to create TiDB Operator CompactBackup CR")
	}
	return deploy.PanicOrDone(err)
}

// PostCreate implements the CompactBackup status monitoring logic
func (t *CompactTask) PostCreate(ctx context.Context) deploy.Result {
	start := time.Now()
	defer func() {
		metricsctrl.ReconcileDuration.WithLabelValues("compact_backup_postcreate").Observe(time.Since(start).Seconds())
	}()

	compactBackup, err := t.getOpCompactBackup(ctx)
	if err != nil {
		t.logger.Error(err, "failed to get TiDB Operator CompactBackup CR")
		t.compactBackup.Status.Phase = cloudv1.CompactPhasePending
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}

	// Always sync the status from the data plane CR to the control plane CR first.
	t.compactBackup.Status.State = cloudv1.DataplaneCompactState(compactBackup.Status.State)
	t.compactBackup.Status.Progress = compactBackup.Status.Progress
	t.compactBackup.Status.Message = compactBackup.Status.Message
	t.compactBackup.Status.EndTs = compactBackup.Status.EndTs
	t.compactBackup.Status.Phase = convertCompactBackupPhase(compactBackup.Status.State)

	// Then, decide the deploy result based on the state.
	switch compactBackup.Status.State {
	case string(cloudv1.DataplaneComplete), string(cloudv1.DataplaneFailed):
		return deploy.Done()
	case string(cloudv1.DataplaneRetryTheFailed):
		// Override the phase for the special retry case.
		t.compactBackup.Status.Phase = cloudv1.CompactPhaseRetry
		return deploy.Done()
	default:
		// For any other state, assume it's still running and needs to be checked again.
		return deploy.Error(fmt.Errorf("compact backup is running, current state: %s", compactBackup.Status.State)).WithRequeuAfter(time.Second * 30)
	}
}

// applyOpCompactBackup creates the CompactBackup CR in the data plane
func (t *CompactTask) applyOpCompactBackup(ctx context.Context, compactBackup *opv1alpha1.CompactBackup) error {
	return k8s.Apply(ctx, t.remote, compactBackup)
}

// getOpCompactBackup retrieves the CompactBackup CR from the data plane
func (t *CompactTask) getOpCompactBackup(ctx context.Context) (*opv1alpha1.CompactBackup, error) {
	key := types.NamespacedName{
		Name:      t.getCompactBackupCRName(),
		Namespace: t.clusterNamer.GetTiDBNamespace(),
	}

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

	return compactBackup, nil
}

// deleteOpCompactBackup deletes the CompactBackup CR from the data plane
func (t *CompactTask) deleteOpCompactBackup(ctx context.Context, compactBackup *opv1alpha1.CompactBackup) error {
	return t.remote.Delete(ctx, compactBackup)
}

// convertCompactBackupPhase converts TiDB Operator CompactBackup state to control plane phase
func convertCompactBackupPhase(state string) cloudv1.CompactPhase {
	switch cloudv1.DataplaneCompactState(state) {
	case cloudv1.DataplaneComplete:
		return cloudv1.CompactPhaseComplete
	case cloudv1.DataplaneRunning:
		return cloudv1.CompactPhaseRunning
	case cloudv1.DataplaneFailed:
		return cloudv1.CompactPhaseFailed
	case cloudv1.DataplaneRetry, cloudv1.DataplaneRetryTheFailed:
		return cloudv1.CompactPhaseRetry
	case cloudv1.DataplaneScheduled, cloudv1.DataplanePrepare:
		return cloudv1.CompactPhasePending
	default:
		return cloudv1.CompactPhasePending
	}
}

// Delete implements the CompactBackup deletion logic
func (t *CompactTask) Delete(ctx context.Context) deploy.Result {
	if t.compactBackup.Status.Phase == cloudv1.CompactPhaseFailed {
		// Ignore deleting dataplane CompactBackup CR for troubleshooting when compact failed
		return deploy.Done()
	}

	compactBackup := &opv1alpha1.CompactBackup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.getCompactBackupCRName(),
			Namespace: t.clusterNamer.GetTiDBNamespace(),
		},
	}

	err := client.IgnoreNotFound(t.deleteOpCompactBackup(ctx, compactBackup))
	if err != nil {
		t.logger.Error(err, "failed to delete TiDB Operator CompactBackup CR")
	}
	return deploy.ErrorOrDone(err)
}

func (t *CompactTask) getPhysicalFileCacheCapacity(ctx context.Context, instanceType string) string {
	logger := log.FromContext(ctx)
	if t.config == nil {
		return ""
	}

	instanceSpec, err := t.config.FindInstanceType(instanceType)
	if err != nil {
		logger.Error(err, "failed to find instance type for physical file cache capacity", "instanceType", instanceType)
		return ""
	}

	availableMemoryBytes := instanceSpec.Memory.Value()
	availableMemoryGiB := availableMemoryBytes / consts.Gi
	if availableMemoryGiB < physicalFileCacheMinAvailableMemoryGiB {
		logger.Info("available memory is below the threshold for physical file cache, skipping configuration",
			"instanceType", instanceType,
			"availableMemoryGiB", availableMemoryGiB,
			"thresholdGiB", physicalFileCacheMinAvailableMemoryGiB)
		return ""
	}

	return fmt.Sprintf("%dG", availableMemoryGiB-physicalFileCacheReservedMemoryGiB)
}