// Package common contains cloud-agnostic helpers for the upgradeoperation controller.
//
// CDCUpgrader provides shared logic for upgrading TiCDC clusters via the ticdc-service
// API. The flow is identical across cloud providers (EKS / GKE / AKS), only the dataplane
// k8s client differs, which is fetched from `dataplane.EnvProvider.GetClient()`.
package common

import (
	"context"
	"fmt"
	"sort"
	"time"

	"github.com/go-logr/logr"
	"github.com/samber/lo"
	"go.uber.org/multierr"
	corev1 "k8s.io/api/core/v1"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/eks-provider/controllers/utils/ticdcservicecli"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	ticdcpb "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
)

// CDCUpgrader drives TiCDC node-version upgrades by calling the ticdc-service API and
// waiting for dataplane pods to reschedule onto new-version nodes. The implementation is
// cloud-agnostic — provider-specific upgraders embed/compose this struct and forward
// `UpgradeCDCNodes` calls to `Upgrade`.
type CDCUpgrader struct {
	infraCli client.Client
	ticdcCli ticdcservicecli.Client
}

// NewCDCUpgrader constructs a CDCUpgrader.
func NewCDCUpgrader(infraCli client.Client, ticdcCli ticdcservicecli.Client) *CDCUpgrader {
	return &CDCUpgrader{
		infraCli: infraCli,
		ticdcCli: ticdcCli,
	}
}

// Upgrade upgrades TiCDC nodes by calling the ticdc-service API to update the
// nodeVersion field on each TiCDC cluster, then waiting for the CDC controller to bring
// them back to Running and verifying that dataplane pods are rescheduled onto the
// new-version nodes.
//
// Whether the TiCDC clusters are upgraded in parallel or serially is controlled by the
// ParallelCDCUpgradeAnnotation step option ("tidbcloud.com/parallel-cdc-upgrade").
// Set it to "true" for parallel; the default is serial.
func (u *CDCUpgrader) Upgrade(ctx context.Context, uo *cloudv1.UpgradeOperation, envProvider dataplane.EnvProvider, cluster *cloudv1.Cluster) (ctrl.Result, error) {
	logger := log.FromContext(ctx).WithValues("cluster", cluster.Name, "desiredVersion", uo.Status.CurrentStep.Version)
	logger.Info("UpgradeCDCNodes")

	desiredVersion := uo.Status.CurrentStep.Version
	parallel := uo.GetCurrentStepOption(ParallelCDCUpgradeAnnotation) == "true"

	var ticdcClusterName []string
	if targetCDCName := uo.GetCurrentStepOption(TargetCDCUpgradeAnnotation); targetCDCName != "" {
		ticdcClusterName = append(ticdcClusterName, targetCDCName)
	}

	cdcListResp, err := u.ticdcCli.ListTiCDCClusters(ctx, cluster.Name, ticdcClusterName)
	if err != nil {
		logger.Error(err, "failed to list TiCDC clusters for the cluster", "ticdcClusterName", ticdcClusterName)
		return ctrl.Result{}, fmt.Errorf("failed to list TiCDC clusters for cluster %s: %w", cluster.Name, err)
	}
	logger.Info("found TiCDC clusters for the cluster", "count", len(cdcListResp.TicdcClusters))

	cdcList := lo.Filter(cdcListResp.TicdcClusters, func(cdc *ticdcpb.TiCDCCluster, _ int) bool {
		higher, err := isNodeVersionHigher(cdc.NodeVersion, desiredVersion)
		if err != nil {
			logger.Error(err, "failed to compare TiCDC cluster nodeVersion with desiredVersion, including cluster for upgrade",
				"ticdcClusterId", cdc.Id, "nodeVersion", cdc.NodeVersion)
		} else if higher {
			logger.Info("TiCDC cluster nodeVersion is higher than desiredVersion, skipping", "ticdcClusterId", cdc.Id, "nodeVersion", cdc.NodeVersion)
			return false
		}
		return cdc.Phase != ticdcpb.TiCDCCluster_DELETING
	})
	if len(cdcList) == 0 {
		logger.Info("no TiCDC clusters match after filtering, finishing CDC node upgrade")
		return ctrl.Result{}, FinishUpgrading(ctx, u.infraCli, uo)
	}

	sort.Slice(cdcList, func(i, j int) bool {
		return cdcList[i].Id < cdcList[j].Id
	})
	logger.Info("upgrading TiCDC clusters for the cluster", "count", len(cdcList), "parallel", parallel)
	namespace := cluster.PrefixedId()

	dpCli := envProvider.GetClient()
	if parallel {
		return u.upgradeCDCsParallel(ctx, uo, dpCli, logger, namespace, cdcList, desiredVersion)
	}
	return u.upgradeCDCsSerial(ctx, uo, dpCli, logger, namespace, cdcList, desiredVersion)
}

// upgradeCDCsParallel triggers a nodeVersion update on all TiCDC clusters at once, then on every
// subsequent reconcile checks that every cluster has returned to Running and its pods are rescheduled.
func (u *CDCUpgrader) upgradeCDCsParallel(
	ctx context.Context,
	uo *cloudv1.UpgradeOperation,
	dpCli client.Client,
	logger logr.Logger,
	namespace string,
	cdcs []*ticdcpb.TiCDCCluster,
	desiredVersion string,
) (ctrl.Result, error) {

	var (
		allReady = true
		errs     error
		done     int
	)

	logger.Info("upgrading all TiCDC clusters in parallel")
	for _, cdc := range cdcs {
		cdcLogger := logger.WithValues("ticdc", cdc.Id, "currentPhase", cdc.Phase, "currentVersion", cdc.NodeVersion)
		cdcLogger.Info("checking TiCDC cluster upgrade status")

		// Step 1: ensure nodeVersion is set to the desired value via API.
		if updated, err := u.ensureCDCNodeVersion(ctx, cdcLogger, cdc, desiredVersion); err != nil || updated {
			allReady = false
			if err != nil {
				errs = multierr.Append(errs, err)
				cdcLogger.Error(err, "failed to update TiCDC cluster nodeVersion")
			} else {
				cdcLogger.Info("updated TiCDC cluster nodeVersion to desired version, waiting for pods to reschedule")
			}
			continue
		}

		// Step 2: verify dataplane pods are on the new-version nodes.
		if rescheduled, err := cdcPodsRescheduled(ctx, dpCli, namespace, cdc.Id, desiredVersion); err != nil || !rescheduled {
			allReady = false
			if err != nil {
				errs = multierr.Append(errs, err)
				cdcLogger.Error(err, "failed to check if TiCDC pods are rescheduled to desiredVersion nodes")
			} else if !rescheduled {
				cdcLogger.Info("TiCDC pods not yet rescheduled to desiredVersion nodes")
			}
			continue
		}

		// Step 3: wait for the TiCDC cluster to return to Running.
		if cdc.Phase != ticdcpb.TiCDCCluster_RUNNING {
			cdcLogger.Info("TiCDC cluster is not Running yet")
			allReady = false
			continue
		}
		done++
	}

	if errs != nil {
		return ctrl.Result{}, errs
	}
	_ = TrackProgressAndUpdate(ctx, u.infraCli, uo, PhaseUpgradeCDC, done, len(cdcs), time.Now())
	if !allReady {
		return ctrl.Result{RequeueAfter: 1 * time.Minute}, nil
	}

	logger.Info("all TiCDC clusters upgraded and pods rescheduled to desiredVersion nodes")
	return ctrl.Result{}, FinishUpgrading(ctx, u.infraCli, uo)
}

// upgradeCDCsSerial upgrades TiCDC clusters one at a time. It will not advance to the next
// cluster until the current one is Running and its pods are confirmed on the new-version nodes.
func (u *CDCUpgrader) upgradeCDCsSerial(
	ctx context.Context,
	uo *cloudv1.UpgradeOperation,
	dpCli client.Client,
	logger logr.Logger,
	namespace string,
	cdcs []*ticdcpb.TiCDCCluster,
	desiredVersion string,
) (ctrl.Result, error) {

	logger.Info("upgrading TiCDC clusters one by one")
	for idx, cdc := range cdcs {
		cdcLogger := logger.WithValues("ticdc", cdc.Id, "currentPhase", cdc.Phase, "currentVersion", cdc.NodeVersion)
		cdcLogger.Info("checking TiCDC cluster upgrade status")

		// Step 1: ensure nodeVersion is set to the desired value via API.
		if updated, err := u.ensureCDCNodeVersion(ctx, cdcLogger, cdc, desiredVersion); err != nil || updated {
			if err != nil {
				cdcLogger.Error(err, "failed to update TiCDC cluster nodeVersion")
			} else {
				cdcLogger.Info("updated TiCDC cluster nodeVersion to desiredVersion, waiting for pods to reschedule")
			}
			_ = TrackProgressAndUpdate(ctx, u.infraCli, uo, PhaseUpgradeCDC, idx, len(cdcs), time.Now())
			return ctrl.Result{Requeue: err == nil}, err
		}

		// Step 2: wait for dataplane pods to land on the new-version nodes.
		if rescheduled, err := cdcPodsRescheduled(ctx, dpCli, namespace, cdc.Id, desiredVersion); err != nil || !rescheduled {
			if err != nil {
				cdcLogger.Error(err, "failed to check if TiCDC pods are rescheduled to desiredVersion nodes")
			} else {
				cdcLogger.Info("waiting for TiCDC pods to be rescheduled to desiredVersion nodes")
			}
			_ = TrackProgressAndUpdate(ctx, u.infraCli, uo, PhaseUpgradeCDC, idx, len(cdcs), time.Now())
			return ctrl.Result{Requeue: err == nil}, err
		}

		// Step 3: wait for the TiCDC cluster to return to Running before moving on.
		if cdc.Phase != ticdcpb.TiCDCCluster_RUNNING {
			cdcLogger.Info("waiting for TiCDC cluster to be Running")
			_ = TrackProgressAndUpdate(ctx, u.infraCli, uo, PhaseUpgradeCDC, idx, len(cdcs), time.Now())
			return ctrl.Result{RequeueAfter: 1 * time.Minute}, nil
		}

		cdcLogger.Info("TiCDC cluster upgraded and pods rescheduled")
	}

	logger.Info("all TiCDC clusters upgraded and pods rescheduled to desiredVersion nodes")
	_ = TrackProgressAndUpdate(ctx, u.infraCli, uo, PhaseUpgradeCDC, len(cdcs), len(cdcs), time.Now())
	return ctrl.Result{}, FinishUpgrading(ctx, u.infraCli, uo)
}

// ensureCDCNodeVersion calls the ticdc-service API to update nodeVersion if it differs from desiredVersion.
// Returns (true, nil) if an update was triggered, (false, nil) if already at the desired version.
func (u *CDCUpgrader) ensureCDCNodeVersion(ctx context.Context, cdcLogger logr.Logger, cdc *ticdcpb.TiCDCCluster, desiredVersion string) (bool, error) {
	if cdc.NodeVersion == desiredVersion {
		return false, nil
	}

	cdcLogger.Info("updating TiCDC cluster nodeVersion via API")

	if err := u.ticdcCli.UpdateTiCDCNodeVersion(ctx, cdc.Id, desiredVersion); err != nil {
		return false, fmt.Errorf("failed to update TiCDC %s node version: %w", cdc.Id, err)
	}
	return true, nil
}

// cdcPodsRescheduled checks that every CDC pod for the given TiCDC cluster in the dataplane
// is scheduled onto a node whose k8s version label matches desiredVersion.
func cdcPodsRescheduled(ctx context.Context, dpCli client.Client, namespace, cdcName, desiredVersion string) (bool, error) {
	pods := &corev1.PodList{}
	if err := dpCli.List(ctx, pods,
		client.InNamespace(namespace),
		client.MatchingLabels{
			"app.kubernetes.io/instance":  cdcName,
			"app.kubernetes.io/component": string(cloudv1.ComponentTiCDC),
		},
	); err != nil {
		return false, fmt.Errorf("failed to list CDC pods: %w", err)
	}
	if len(pods.Items) == 0 {
		return false, nil
	}
	for _, pod := range pods.Items {
		if pod.Spec.NodeName == "" {
			return false, nil
		}
		if pod.Spec.NodeSelector[consts.K8sVersionLabelKey] != desiredVersion {
			return false, nil
		}
	}
	return true, nil
}