package nodepoolgc

import (
	"context"
	stderrors "errors"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/prometheus/client_golang/prometheus"

	"github.com/go-logr/logr"
	"github.com/pingcap/errors"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/metrics"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	regionalv1alpha1 "github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/metrics/collectors"
)

const (
	controllerName      = "br_nodepool_gc"
	defaultScanInterval = time.Hour
	defaultExpireAfter  = 7 * 24 * time.Hour
)

// errAPPBoxNotFound identifies the safe-to-classify APPBox resolution failure
// used by the local orphan cleanup path.
var errAPPBoxNotFound = stderrors.New("APPBox not found")

type appBoxNotFoundError struct {
	provider       string
	kubernetesName string
}

func (e *appBoxNotFoundError) Error() string {
	return fmt.Sprintf(
		"no APPBox found for provider=%s kubernetes=%s",
		e.provider,
		e.kubernetesName,
	)
}

func (e *appBoxNotFoundError) Unwrap() error {
	return errAPPBoxNotFound
}

type Status string

type failureReason string

const (
	failureReasonUnknown      failureReason = "unknown"
	failureReasonHavingNodes  failureReason = "havingNodes"
	failureReasonFailToDelete failureReason = "failToDelete"
)

const (
	StatusNotExpired       Status = "not_expired"
	StatusExpired          Status = "expired"
	StatusDeletionApproved Status = "deletion_approved"
	StatusDeleted          Status = "deleted"
	StatusFailed           Status = "failed"
)

var statusValueMap = map[Status]float64{
	StatusNotExpired:       0,
	StatusExpired:          1,
	StatusDeletionApproved: 2,
	StatusDeleted:          3,
	StatusFailed:           4,
}

// nodePool is the smallest common representation needed by the GC.
type nodePool struct {
	// primary is a NodeGroup on AWS and a NodePool on GCP or Azure.
	primary client.Object
	// companions are other objects(e.g. LaunchTemplates on AWS) that should be deleted along with the primary object.
	companions []client.Object
	provider   string
	region     string
	// kubernetesName identifies the APPBox that owns this nodepool.
	kubernetesName string
}

func (p nodePool) label(key string) (_ string, ok bool) {
	val, ok := p.primary.GetLabels()[key]
	return val, ok
}

func (p nodePool) component() string {
	val, _ := p.label(namer.ComponentSelectorKey)
	return val
}

func (p nodePool) tenantID() string {
	val, _ := p.label(consts.TagKeyTenant)
	return val
}

func (p nodePool) projectID() string {
	val, _ := p.label(consts.TagKeyProject)
	return val
}

func (p nodePool) clusterID() string {
	val, _ := p.label(consts.TagKeyCluster)
	return val
}

type GC struct {
	rsmanager         *controllers.Manager
	local             client.Client
	logger            logr.Logger
	config            config.NodePoolGCConfig
	providers         []CloudProvider
	dataplaneResolver func(context.Context, nodePool) (client.Client, error)
	now               func() time.Time
}

type processResult struct {
	status Status
	reason failureReason
	skip   bool
}

func New(
	local client.Client,
	logger logr.Logger,
	cfg config.NodePoolGCConfig,
	providerType string,
	rsmanager *controllers.Manager,
) (*GC, error) {
	cfg = configWithDefaults(cfg)
	g := &GC{
		rsmanager: rsmanager,
		local:     local,
		logger:    logger,
		config:    cfg,
		now:       time.Now,
	}

	provider, err := newCloudProvider(providerType, local)
	if err != nil {
		return nil, err
	}
	g.providers = []CloudProvider{provider}
	g.dataplaneResolver = g.remoteClientForPool
	return g, nil
}

func configWithDefaults(cfg config.NodePoolGCConfig) config.NodePoolGCConfig {
	if cfg.ScanInterval.Duration <= 0 {
		cfg.ScanInterval = metav1.Duration{Duration: defaultScanInterval}
	}
	if cfg.ExpireAfter.Duration <= 0 {
		cfg.ExpireAfter = metav1.Duration{Duration: defaultExpireAfter}
	}
	return cfg
}

func (g *GC) Start(ctx context.Context) error {
	ticker := time.NewTicker(g.config.ScanInterval.Duration)
	defer ticker.Stop()

	g.RunOnce(ctx)
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			g.RunOnce(ctx)
		}
	}
}

func (g *GC) RunOnce(ctx context.Context) {
	logger := g.logger.WithValues("round", uuid.New().String())

	ctx = log.IntoContext(ctx, logger)
	for _, provider := range g.providers {
		g.scanProvider(ctx, provider)
	}
}

func (g *GC) scanProvider(ctx context.Context, provider CloudProvider) {
	defer g.recoverPanic(ctx, "")
	logger := log.FromContext(ctx)

	pools, err := provider.ListNodePools(ctx)
	if err != nil {
		logger.Error(err, "failed to list BR nodepools", "provider", provider.Name())
		return
	}
	for _, pool := range pools {
		// Skip pools that explicitly opt out of node pool GC management.
		if managed, exists := pool.label(consts.LabelKeyNodePoolGCManaged); exists && managed != consts.AnnoValTrue {
			continue
		}

		result := g.processPool(ctx, provider, pool)
		if !result.skip {
			g.recordStatus(pool, result.status, result.reason)
		}
	}
}

func (g *GC) processPool(ctx context.Context, provider CloudProvider, pool nodePool) processResult {
	defer g.recoverPanic(ctx, pool.region)
	logger := log.FromContext(ctx)

	original := pool.primary.DeepCopyObject().(client.Object)
	status, stateErr := updateGCMetadata(g.now().UTC(), pool.primary, g.config)
	if stateErr != nil {
		logger.Error(stateErr, "failed to evaluate BR nodepool GC state", "name", pool.primary.GetName())
		return processResult{status: StatusFailed, reason: failureReasonUnknown}
	}
	logger.Info("nodepool GC state", "name", pool.primary.GetName(), "status", status)
	if g.config.DryRun {
		logger.Info("dry-run enabled; skipping BR nodepool GC metadata patch, dataplane Node check, and deletion", "name", pool.primary.GetName())
		return processResult{status: status}
	}
	if err := g.local.Patch(ctx, pool.primary, client.MergeFrom(original)); err != nil {
		logger.Error(err, "failed to patch BR nodepool GC metadata", "name", pool.primary.GetName())
		return processResult{status: StatusFailed, reason: failureReasonUnknown}
	}
	if status != StatusDeletionApproved {
		return processResult{status: status}
	}

	hasNodes, err := g.hasMatchingNodes(ctx, provider, pool)
	if err != nil {
		if stderrors.Is(err, errAPPBoxNotFound) {
			// APPBox teardown deletes the cloud cluster (EKS/GKE/AKS) and its
			// managed node pools before the APPBox record itself disappears, so
			// an absent APPBox implies the underlying cloud resources are
			// already gone. Deleting the local MRs is therefore a no-op on the
			// cloud side even when they still carry finalizers: the Crossplane
			// provider observes the external resource is missing and releases
			// its finalizer, letting the record be cleaned up.
			logger.Info("APPBox is absent; deleting local orphan records", "name", pool.primary.GetName(), "hasFinalizers", poolHasFinalizers(pool))
		} else if isDataplaneNotReady(err) {
			logger.Info("dataplane dependency is not ready; skipping this nodepool until the next scan", "name", pool.primary.GetName(), "error", err)
			return processResult{skip: true}
		} else {
			logger.Error(err, "failed to double check BR nodepool nodes", "name", pool.primary.GetName())
			return processResult{status: StatusFailed, reason: failureReasonUnknown}
		}
	} else if hasNodes {
		err := fmt.Errorf("an expired nodegroup still has running nodes, name=%s", pool.primary.GetName())
		logger.Error(err, "skipping deletion")
		return processResult{status: StatusFailed, reason: failureReasonHavingNodes}
	}

	if err := g.deletePool(ctx, pool); err != nil {
		logger.Error(err, "failed to delete BR nodepool", "name", pool.primary.GetName())
		return processResult{status: StatusFailed, reason: failureReasonFailToDelete}
	}
	return processResult{status: StatusDeleted}
}

func poolHasFinalizers(pool nodePool) bool {
	if len(pool.primary.GetFinalizers()) != 0 {
		return true
	}
	for _, companion := range pool.companions {
		if len(companion.GetFinalizers()) != 0 {
			return true
		}
	}
	return false
}

func (g *GC) deletePool(ctx context.Context, pool nodePool) error {
	logger := log.FromContext(ctx)

	for _, companion := range pool.companions {
		logger.Info("deleting companion resource", "name", companion.GetName(), "kind", companion.GetObjectKind().GroupVersionKind().Kind)
		if err := client.IgnoreNotFound(g.local.Delete(ctx, companion)); err != nil {
			return err
		}
	}
	logger.Info("deleting nodepool", "name", pool.primary.GetName(), "kind", pool.primary.GetObjectKind().GroupVersionKind().Kind)
	return client.IgnoreNotFound(g.local.Delete(ctx, pool.primary))
}

func (g *GC) hasMatchingNodes(ctx context.Context, provider CloudProvider, pool nodePool) (bool, error) {
	remote, err := g.dataplaneResolver(ctx, pool)
	if err != nil {
		return false, err
	}

	var nodes corev1.NodeList
	if err := remote.List(ctx, &nodes, client.MatchingLabels(provider.NodeSelector(pool))); err != nil {
		return false, err
	}
	return len(nodes.Items) > 0, nil
}

func isDataplaneNotReady(err error) bool {
	var unregistered *controllers.APPBoxUnregisteredError
	if stderrors.As(err, &unregistered) {
		return true
	}
	var cacheNotStarted *cache.ErrCacheNotStarted
	return stderrors.As(err, &cacheNotStarted)
}

func (g *GC) recordStatus(pool nodePool, current Status, reason failureReason) {
	if current != StatusFailed {
		reason = ""
	}
	if current == StatusFailed && reason == "" {
		reason = failureReasonUnknown
	}
	labels := []string{
		pool.provider,
		pool.region,
		pool.component(),
		pool.primary.GetNamespace(),
		pool.primary.GetName(),
		pool.tenantID(),
		pool.projectID(),
		pool.clusterID(),
		string(reason),
	}
	collectors.NodePoolGCStatus.DeletePartialMatch(prometheus.Labels{"namespace": pool.primary.GetNamespace(), "name": pool.primary.GetName()})
	collectors.NodePoolGCStatus.WithLabelValues(labels...).Set(statusValueMap[current])
}

func (g *GC) recoverPanic(ctx context.Context, region string) {
	if recovered := recover(); recovered != nil {
		metrics.ControllerPanic.WithLabelValues(controllerName, region).Inc()
		log.FromContext(ctx).Error(errors.Errorf("panic in BR nodepool GC: %v", recovered), "recovered from panic", "region", region)
	}
}

func (g *GC) remoteClientForPool(ctx context.Context, pool nodePool) (client.Client, error) {
	if pool.kubernetesName == "" {
		return nil, errors.Errorf("nodepool %s has no Kubernetes cluster reference", pool.primary.GetName())
	}

	var appBoxes regionalv1alpha1.APPBoxList
	if err := g.local.List(ctx, &appBoxes, appBoxSelectorForPool(pool)); err != nil {
		return nil, errors.Wrap(err, "cannot list APPBoxes for nodepool GC")
	}
	for i := range appBoxes.Items {
		appBox := &appBoxes.Items[i]
		if appBox.Spec.Provider != pool.provider || appBox.Status.KubernetesClusterName != pool.kubernetesName {
			continue
		}
		getter, err := g.rsmanager.ClientGetter(ctx, client.ObjectKeyFromObject(appBox))
		if err != nil {
			return nil, err
		}
		return getter.GetClient(ctx)
	}

	return nil, &appBoxNotFoundError{provider: pool.provider, kubernetesName: pool.kubernetesName}
}

func appBoxSelectorForPool(pool nodePool) client.MatchingLabels {
	selector := client.MatchingLabels{
		cloudv1.LabelServiceType:       consts.ServiceTypeDedicated,
		regionalv1alpha1.TagTenantKey:  pool.tenantID(),
		regionalv1alpha1.TagProjectKey: pool.projectID(),
	}
	return selector
}