package workflow

import (
	"context"
	"fmt"
	"reflect"
	"strings"
	"time"

	"github.com/go-logr/logr"
	"github.com/google/uuid"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	"github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	"github.com/tidbcloud/infra-provider/pkg/consts"

	"github.com/tidbcloud/eks-provider/controllers/utils/configmanager"
	"github.com/tidbcloud/eks-provider/controllers/workflow/action"
	"github.com/tidbcloud/eks-provider/controllers/workflow/action/az"
	_ "github.com/tidbcloud/eks-provider/controllers/workflow/action/basic"
	"github.com/tidbcloud/eks-provider/controllers/workflow/action/br"
	"github.com/tidbcloud/eks-provider/controllers/workflow/action/cluster"
	"github.com/tidbcloud/eks-provider/controllers/workflow/action/common"
	dataplaneAction "github.com/tidbcloud/eks-provider/controllers/workflow/action/dataplane"
	"github.com/tidbcloud/eks-provider/controllers/workflow/action/dualdisk"
	"github.com/tidbcloud/eks-provider/controllers/workflow/action/illegalboundary"
	"github.com/tidbcloud/eks-provider/controllers/workflow/action/vpcLogs"
	"github.com/tidbcloud/eks-provider/controllers/workflow/action/zonalshift"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	metricsctrl "github.com/tidbcloud/infra-provider/pkg/metrics/controller"
	utiltime "github.com/tidbcloud/infra-provider/pkg/util/time"
)

const (
	Finalizer = "workflow.cloud.pingcap.com/finalizer"
)

func (r *WorkflowReconciler) shouldWaitForNextExecuteTime(wf *cloudv1.Workflow) (time.Duration, bool) {
	if wf.Status.NextExecuteTime == nil {
		return 0, false
	}
	now := r.clock.Now()
	if duration := wf.Status.NextExecuteTime.Time.Sub(now); duration > 0 {
		return duration, true
	}
	return 0, false
}

func (r *WorkflowReconciler) calculateTimeToDelete(wf *cloudv1.Workflow) time.Time {
	ttl := wf.Spec.TTL
	if ttl == nil {
		// Fallback to default TTL if not set
		ttl = &metav1.Duration{Duration: 365 * 24 * time.Hour}
	}
	if wf.Status.EndAt != nil {
		return wf.Status.EndAt.Add(ttl.Duration)
	}
	if wf.Status.BeginAt != nil {
		return wf.Status.BeginAt.Add(ttl.Duration)
	}
	return wf.GetCreationTimestamp().Add(ttl.Duration)
}

// WorkflowReconciler reconciles a ClusterConfig object
type WorkflowReconciler struct {
	client            client.Client
	eventRecorder     record.EventRecorder
	registry          *action.Registry
	scheduler         Scheduler
	cachedK8sProvider dataplane.EnvProviderManager

	GCPCli gcp.Client
	AWSCli aws.Client

	// AssumeRoleName / AssumeRoleExternalID are used by AWS workflow actions
	// to assume into dataplane accounts when Network Spec has no role ARN.
	AssumeRoleName       string
	AssumeRoleExternalID string

	// PDAuthFunc authenticates HTTP requests to the dataplane cluster proxy / PD API (same source as cluster.ClusterReconciler).
	PDAuthFunc        configmanager.PDAuth
	PDServiceResolver cluster.PDServiceResolver
	DryRun            bool

	clock utiltime.Clock
	Log   logr.Logger
}

type Option func(r *WorkflowReconciler)

// WithPDAuth sets the PD/dataplane proxy auth callback used by workflow actions that call PD through the proxy.
func WithPDAuth(auth configmanager.PDAuth) Option {
	return func(r *WorkflowReconciler) {
		r.PDAuthFunc = auth
	}
}

func WithPDServiceResolver(pdServiceResolver cluster.PDServiceResolver) Option {
	return func(r *WorkflowReconciler) {
		r.PDServiceResolver = pdServiceResolver
	}
}

func WithDryRun(dryRun bool) Option {
	return func(r *WorkflowReconciler) {
		r.DryRun = dryRun
	}
}

func WithAWSCli(awsCli aws.Client) Option {
	return func(r *WorkflowReconciler) {
		r.AWSCli = awsCli
	}
}

func WithAWSAssumeRole(assumeRoleName, externalID string) Option {
	return func(r *WorkflowReconciler) {
		r.AssumeRoleName = assumeRoleName
		r.AssumeRoleExternalID = externalID
	}
}

func registerActions(r *WorkflowReconciler) {
	action.RegisterWorkflowAction(common.NewPauseReconciliation(r.client))
	action.RegisterWorkflowAction(common.NewResumeReconciliation(r.client))
	action.RegisterWorkflowAction(dataplaneAction.NewDeleteTCMember(r.client))
	action.RegisterWorkflowAction(dataplaneAction.NewRestartPod(r.client))
	action.RegisterWorkflowAction(dataplaneAction.NewRestartComponent(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(cluster.NewSetStandbyNode(r.client))
	action.RegisterWorkflowAction(dataplaneAction.NewCordonNode(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(dataplaneAction.NewWaitPodPhase(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(dataplaneAction.NewGetPodInfo(r.client))
	action.RegisterWorkflowAction(dataplaneAction.NewTogglePDScheduler(r.client))
	action.RegisterWorkflowAction(br.NewCreateNodePool(r.client))
	action.RegisterWorkflowAction(br.NewDeleteNodePool(r.client))
	action.RegisterWorkflowAction(br.NewCreateBackup(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(br.NewDeleteBackup(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(br.NewCreateRestore(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(br.NewDeleteRestore(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(cluster.NewDeleteTiDBClusterNodes(r.client))
	action.RegisterWorkflowAction(dataplaneAction.NewDeleteASTS(r.client))
	action.RegisterWorkflowAction(cluster.NewUpdateClusterStatus(r.client))
	action.RegisterWorkflowAction(dataplaneAction.NewRestartOtherPod(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(dualdisk.NewRaftMigrationUpdateCluster(r.client))
	action.RegisterWorkflowAction(dualdisk.NewEnsureDataplaneConfigMapContain(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(dualdisk.NewDeleteOneASTS(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(dualdisk.NewEnsureASTSContainVolume(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(dualdisk.NewEnsureAstsAdoptPods(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(dualdisk.NewCreatePVCs(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(dualdisk.NewRaftMigrationCopyRaft(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(dataplaneAction.NewCheckPodStatus(r.client))
	action.RegisterWorkflowAction(illegalboundary.NewCheckTFRegionKeys(r.client))
	action.RegisterWorkflowAction(illegalboundary.NewMergeTFRegionKeys(r.client))
	action.RegisterWorkflowAction(illegalboundary.NewRepairInvalidLearner(r.client))
	action.RegisterWorkflowAction(vpcLogs.NewModifyLoadBalanceLogs(r.client, r.GCPCli, r.cachedK8sProvider))
	action.RegisterWorkflowAction(vpcLogs.NewModifyVPCFlowLogs(r.client, r.GCPCli, r.cachedK8sProvider))
	action.RegisterWorkflowAction(az.NewOfflineAZ(r.client, r.cachedK8sProvider, r.AWSCli, r.AssumeRoleName, r.AssumeRoleExternalID))
	action.RegisterWorkflowAction(az.NewRecoverAZ(r.client, r.cachedK8sProvider, r.AWSCli, r.AssumeRoleName, r.AssumeRoleExternalID))

	// cluster related actions
	action.RegisterWorkflowAction(cluster.NewToggleLBTargetAction(r.client))
	action.RegisterWorkflowAction(cluster.NewWaitClusterSyncedAction(r.client))
	action.RegisterWorkflowAction(cluster.NewRemoveTiProxyAction(r.client))
	action.RegisterWorkflowAction(cluster.NewScaleClusterAction(r.client, r.cachedK8sProvider, r.PDAuthFunc, r.PDServiceResolver, r.DryRun))
	action.RegisterWorkflowAction(common.NewMutateAnnotationsAction(r.client))
	action.RegisterWorkflowAction(common.NewEnsureAnnotateToTCAction(r.client, r.cachedK8sProvider))
	action.RegisterWorkflowAction(common.NewWaitTargetReady(r.client, r.cachedK8sProvider))

	if r.AWSCli != nil {
		zsAPI := zonalshift.NewSDKAWSAPI(r.AWSCli)
		action.RegisterWorkflowAction(zonalshift.NewStartZonalShift(r.client, zsAPI, r.cachedK8sProvider, r.AssumeRoleName, r.AssumeRoleExternalID))
		action.RegisterWorkflowAction(zonalshift.NewUpdateZonalShift(r.client, zsAPI, r.cachedK8sProvider, r.AssumeRoleName, r.AssumeRoleExternalID))
		action.RegisterWorkflowAction(zonalshift.NewCancelZonalShift(r.client, zsAPI, r.cachedK8sProvider, r.AssumeRoleName, r.AssumeRoleExternalID))
		action.RegisterWorkflowAction(zonalshift.NewCleanupZonalShift(r.client, zsAPI, r.cachedK8sProvider, r.AssumeRoleName, r.AssumeRoleExternalID))
	}
}

func NewWorkflowReconciler(mgr ctrl.Manager, cachedProvider dataplane.EnvProviderManager, log logr.Logger, clock utiltime.Clock, gcpCli gcp.Client, opts ...Option) *WorkflowReconciler {
	cli := mgr.GetClient()
	r := &WorkflowReconciler{
		client:            cli,
		eventRecorder:     mgr.GetEventRecorderFor("workflow-controller"),
		registry:          action.DefaultRegistry,
		scheduler:         NewSequenceScheduler(action.DefaultRegistry), // Currently, support only sequence steps
		cachedK8sProvider: cachedProvider,
		clock:             clock,
		Log:               log,
		GCPCli:            gcpCli,
	}
	for _, opt := range opts {
		opt(r)
	}

	registerActions(r)
	return r
}

type ReconcileContext struct {
	Ctx    context.Context
	Logger logr.Logger

	LocalClient client.Client
	Workflow    *cloudv1.Workflow
}

// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=workflows,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=workflows/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=workflows/finalizers,verbs=update
func (r *WorkflowReconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ctrl.Result, err error) {
	start := time.Now()
	logger := log.FromContext(ctx).
		WithValues(
			"reconcile-id", uuid.New().String(),
			"name", req.NamespacedName,
		)
	ctx = log.IntoContext(ctx, logger)
	defer func() {
		metricsctrl.ReconcileDuration.WithLabelValues("workflow").Observe(time.Since(start).Seconds())

		params := []interface{}{"take", time.Since(start)}
		if err == nil {
			params = append(params, "result", result)
		}
		logger.Info("finish reconcile", params...)
	}()

	logger.Info("start reconcile Workflow")
	wf := &cloudv1.Workflow{}
	if err = r.client.Get(ctx, req.NamespacedName, wf); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}
	defer func() {
		if err == nil {
			r.eventRecorder.Event(wf, corev1.EventTypeNormal, "ReconcileSuccess", "")
		} else {
			r.eventRecorder.Event(wf, corev1.EventTypeWarning, "ReconcileFail", err.Error())
		}
	}()

	if wf.Spec.Paused {
		logger.Info("skip, Workflow is paused")
		LogAndEmitEvent(ctx, r.eventRecorder, wf, corev1.EventTypeNormal, "Workflow is paused, will not process it", nil)
		return ctrl.Result{}, nil
	}
	// handle cancel annotation
	if wf.Annotations != nil {
		if _, ok := wf.Annotations[consts.AnnoCancelWorkflow]; ok {
			if err := r.handleCancelAnnotation(ctx, wf); err != nil {
				return ctrl.Result{}, fmt.Errorf("handle cancel annotation: %w", err)
			}
		}
	}

	switch wf.Status.Phase {
	case cloudv1.WorkflowPhaseNone:
		return r.handleNone(ctx, wf)
	case cloudv1.WorkflowPhasePending:
		return r.handlePending(ctx, wf)
	case cloudv1.WorkflowPhaseRunning:
		return r.handleRunning(ctx, wf)
	case cloudv1.WorkflowPhaseCompleted, cloudv1.WorkflowPhaseFailed, cloudv1.WorkflowPhaseCancelled:
		return r.handleWorkflowDone(ctx, wf)
	case cloudv1.WorkflowPhaseCancelling:
		return r.handleCancelling(ctx, wf)
	default:
		logger.Info("unknown phase, ignore it")
		return ctrl.Result{}, nil
	}
}

// SetupWithManager sets up the controller with the Manager.
func (r *WorkflowReconciler) SetupWithManager(mgr ctrl.Manager, pds ...predicate.Predicate) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.Workflow{}, builder.WithPredicates(pds...)).
		WithOptions(controller.Options{RateLimiter: k8s.RetryLimiter}).
		Complete(r)
}

func (r *WorkflowReconciler) handleNone(ctx context.Context, wf *cloudv1.Workflow) (ctrl.Result, error) {
	now := r.clock.Now()

	// set default spec
	wf.Default()
	if err := r.client.Update(ctx, wf); err != nil {
		return ctrl.Result{}, fmt.Errorf("update workflow for defaulting: %w", err)
	}

	if err := wf.Init(ctx, now); err != nil {
		return ctrl.Result{}, fmt.Errorf("init workflow: %w", err)
	}
	for _, each := range wf.Spec.Steps {
		act, ok := r.registry.Actions[each.Uses]
		if !ok {
			return ctrl.Result{}, setWorkflowAsFailed(ctx, r.client, wf, fmt.Sprintf("unknown action refereced by step, action=%s, step=%s", each.Uses, each.Name))
		}

		if err := act.PreCheck(ctx, wf, each); err != nil {
			return ctrl.Result{}, setWorkflowAsFailed(ctx, r.client, wf, fmt.Sprintf("step PreCheck failed, action=%s, step=%s, err=%s", each.Uses, each.Name, err))
		}
	}
	LogAndEmitEvent(ctx, r.eventRecorder, wf, corev1.EventTypeNormal, "workflow is initialized", nil)
	if err := r.client.Status().Update(ctx, wf); err != nil {
		return ctrl.Result{}, err
	}
	return ctrl.Result{}, nil
}

func (r *WorkflowReconciler) handlePending(ctx context.Context, wf *cloudv1.Workflow) (ctrl.Result, error) {
	now := r.clock.Now()
	if wf.Spec.Schedule != nil && now.Before(wf.Spec.Schedule.Time) {
		interval := wf.Spec.Schedule.Sub(now)
		LogAndEmitEvent(ctx, r.eventRecorder, wf, corev1.EventTypeNormal, "Wait for the schedule", nil)
		return ctrl.Result{RequeueAfter: interval}, nil
	}

	if err := wf.Run(ctx, "Running"); err != nil {
		return ctrl.Result{}, fmt.Errorf("run workflow: %w", err)
	}
	LogAndEmitEvent(ctx, r.eventRecorder, wf, corev1.EventTypeNormal, "The schedule is reached", nil)
	if err := r.client.Status().Update(ctx, wf); err != nil {
		return ctrl.Result{}, err
	}
	return ctrl.Result{}, nil
}

// handleCancelling processes a workflow in the Cancelling phase.
// It schedules steps for cancellation, executes running/ending steps, and handles cancellation cleanup.
func (r *WorkflowReconciler) handleCancelling(ctx context.Context, wf *cloudv1.Workflow) (res ctrl.Result, err error) {
	logger := log.FromContext(ctx)
	logger.Info("workflow is cancelling")

	// Check if we should wait for NextExecuteTime
	if duration, shouldWait := r.shouldWaitForNextExecuteTime(wf); shouldWait {
		logger.Info("wait for next execute time", "duration", duration)
		return ctrl.Result{RequeueAfter: duration}, nil
	}

	defer func() {
		if err == nil && wf.Status.Phase == cloudv1.WorkflowPhaseCancelling {
			res, err = r.parseAndSetNextExecuteTime(ctx, wf)
		}
	}()

	scheduled, err := r.scheduler.Schedule(ctx, wf)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("scheduling during cancelling: %w", err)
	}
	if scheduled {
		if err := r.client.Status().Update(ctx, wf); err != nil {
			return ctrl.Result{}, fmt.Errorf("update workflow: %w", err)
		}
	}

	// Execute all step types in order
	if err := r.executeRunningSteps(ctx, wf); err != nil {
		return ctrl.Result{}, err
	}
	if err := r.executeEndingSteps(ctx, wf); err != nil {
		return ctrl.Result{}, err
	}
	if err := r.executeCancellingSteps(ctx, wf); err != nil {
		return ctrl.Result{}, err
	}
	if err := r.executeCancelEndingSteps(ctx, wf); err != nil {
		return ctrl.Result{}, err
	}

	return ctrl.Result{}, nil
}

// executeCancellingSteps executes steps that are in the Cancelling phase.
func (r *WorkflowReconciler) executeCancellingSteps(ctx context.Context, wf *cloudv1.Workflow) error {
	return r.executeSteps(ctx, wf, wf.Status.CancellingSteps, "cancelling")
}

// executeCancelEndingSteps executes steps that are in the CancelEnding phase.
func (r *WorkflowReconciler) executeCancelEndingSteps(ctx context.Context, wf *cloudv1.Workflow) error {
	return r.executeSteps(ctx, wf, wf.Status.CancelEndingSteps, "cancel ending")
}

// isWorkflowInTerminalOrCancellingPhase checks if the workflow is in a terminal or cancelling phase.
func isWorkflowInTerminalOrCancellingPhase(phase cloudv1.WorkflowPhase) bool {
	return phase == cloudv1.WorkflowPhaseCancelling ||
		phase == cloudv1.WorkflowPhaseCancelled ||
		phase == cloudv1.WorkflowPhaseCompleted ||
		phase == cloudv1.WorkflowPhaseFailed
}

// cancelPendingWorkflow cancels a workflow that is in Pending or None phase.
// All steps are immediately marked as Cancelled since they haven't started yet.
func (r *WorkflowReconciler) cancelPendingWorkflow(ctx context.Context, wf *cloudv1.Workflow) error {
	if err := wf.Cancelled(ctx, "canceled"); err != nil {
		return fmt.Errorf("mark workflow as cancelled: %w", err)
	}

	// For Pending/None phase workflows, all steps should be marked as Cancelled
	for _, stepStatus := range wf.Status.Steps {
		if stepStatus == nil {
			continue
		}
		stepStatus.Phase = cloudv1.WorkflowStepPhaseCancelled
		wf.Status.FinishedSteps = append(wf.Status.FinishedSteps, stepStatus.Name)
	}

	return r.client.Status().Update(ctx, wf)
}

// handleCancelAnnotation processes the cancel annotation on a workflow.
// It transitions the workflow to the appropriate cancellation phase based on its current state.
func (r *WorkflowReconciler) handleCancelAnnotation(ctx context.Context, wf *cloudv1.Workflow) error {
	// Skip if workflow is already in a terminal or cancelling phase
	if isWorkflowInTerminalOrCancellingPhase(wf.Status.Phase) {
		return nil
	}

	// For workflows that haven't started, cancel immediately
	if wf.Status.Phase == cloudv1.WorkflowPhasePending || wf.Status.Phase == cloudv1.WorkflowPhaseNone {
		return r.cancelPendingWorkflow(ctx, wf)
	}

	// For running workflows, transition to Cancelling phase
	if err := wf.Cancel(ctx, "cancelling"); err != nil {
		return fmt.Errorf("mark workflow as cancelling: %w", err)
	}

	return r.client.Status().Update(ctx, wf)
}

func (r *WorkflowReconciler) handleRunning(ctx context.Context, wf *cloudv1.Workflow) (res ctrl.Result, err error) {
	// check whether reach the NextExecuteTime
	// adding this logic is used to avoid too many useless reconciliations which caused by the status update such as '.status.steps[].count'
	if duration, shouldWait := r.shouldWaitForNextExecuteTime(wf); shouldWait {
		return ctrl.Result{RequeueAfter: duration}, nil
	}

	defer func() {
		if err == nil && wf.Status.Phase == cloudv1.WorkflowPhaseRunning {
			res, err = r.parseAndSetNextExecuteTime(ctx, wf)
		}
	}()

	scheduled, err := r.scheduler.Schedule(ctx, wf)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("scheduling during running: %w", err)
	}
	if scheduled {
		if err := r.client.Status().Update(ctx, wf); err != nil {
			return ctrl.Result{}, fmt.Errorf("update workflow: %w", err)
		}
	}

	if err = r.executeRunningSteps(ctx, wf); err != nil {
		return ctrl.Result{}, err
	}

	if err = r.executeEndingSteps(ctx, wf); err != nil {
		return ctrl.Result{}, err
	}

	// if all running steps are finished in current turn, there should be some status changed and then the next reconciliation should be
	// triggered immediately after current this return.
	return ctrl.Result{}, nil
}

// executeSteps executes steps from the given step list with a descriptive name for error messages
func (r *WorkflowReconciler) executeSteps(ctx context.Context, wf *cloudv1.Workflow, stepNames []string, stepType string) error {
	logger := log.FromContext(ctx)
	for _, stepName := range stepNames {
		err := r.executeOneStep(ctx, wf, stepName)
		if err != nil {
			logger.Error(err, fmt.Sprintf("execute step %s failed", stepType), "step", stepName)
			return fmt.Errorf("execute step %s failed: %w", stepType, err)
		}

		if err := r.client.Status().Update(ctx, wf); err != nil {
			logger.Error(err, "update workflow status failed", "step", stepName)
			return err
		}
	}
	return nil
}

func (r *WorkflowReconciler) executeRunningSteps(ctx context.Context, wf *cloudv1.Workflow) error {
	return r.executeSteps(ctx, wf, wf.Status.RunningSteps, "running")
}

func (r *WorkflowReconciler) executeEndingSteps(ctx context.Context, wf *cloudv1.Workflow) error {
	return r.executeSteps(ctx, wf, wf.Status.EndingSteps, "ending")
}

func (r *WorkflowReconciler) executeOneStep(ctx context.Context, wf *cloudv1.Workflow, stepName string) error {
	logger := log.FromContext(ctx)
	if wf.Done() {
		logger.Info("skip execute step as the workflow is done", "step", stepName, "phase", wf.Status.Phase)
		return nil
	}

	stepSpec, ok := wf.Step(ctx, stepName)
	if !ok {
		return fmt.Errorf("unknown step: %s", stepName)
	}
	act, ok := r.registry.Actions[stepSpec.Uses]
	if !ok {
		return fmt.Errorf("unknown action refereced by step, action=%s, step=%s", stepSpec.Uses, stepName)
	}

	// inflate step's inputs
	inflated, err := inflateStepInputs(ctx, wf, stepSpec.Inputs.Object)
	if err != nil {
		return fmt.Errorf("inflate step(%s) inputs: %w", stepName, err)
	}
	// Note:
	//   Since the stepSpec is a value rather than a pointer, so that this is not impacted on the original spec.
	// On the other side, we do only cli.Status().Update() in WF reconciliation loop, so the step spec will not be updated.
	stepSpec.Inputs.Object = inflated

	stepStatus, ok := wf.StepStatus(ctx, stepName)
	if !ok || stepStatus == nil {
		return fmt.Errorf("step status not found: %s", stepName)
	}
	switch phase := stepStatus.Phase; phase {
	case cloudv1.WorkflowStepPhaseRunning:
		return wf.OnStepRunningResult(ctx, stepName, act.Handle(ctx, wf, stepSpec))
	case cloudv1.WorkflowStepPhaseCancelling:
		actCancel, ok := act.(action.ActionCancel)
		if ok {
			return wf.OnStepCancellingResult(ctx, stepName, actCancel.Cancel(ctx, wf, stepSpec, false))
		}
		return fmt.Errorf("action supports cancellation, but step %s does not support cancellation", stepName)

	case cloudv1.WorkflowStepPhaseEnding:
		var endingErr error
		if actPost, ok := act.(action.ActionPost); ok {
			endingErr = actPost.PostHandle(ctx, wf, stepSpec)
		}
		return wf.OnStepEndingResult(ctx, stepName, endingErr)
	case cloudv1.WorkflowStepPhaseCancelEnding:
		actCancel, ok := act.(action.ActionCancel)
		if ok {
			return wf.OnStepCancelEndingResult(ctx, stepName, actCancel.PostCancel(ctx, wf, stepSpec))
		}
		return fmt.Errorf("action supports cancellation, but step %s does not support cancellation", stepName)

	default:
		return fmt.Errorf("step is not in Running, Ending, Cancelling, or Cancelled phase, name=%s, phase=%s", stepName, phase)
	}
}

func (r *WorkflowReconciler) parseAndSetNextExecuteTime(ctx context.Context, wf *cloudv1.Workflow) (ctrl.Result, error) {
	now := r.clock.Now()

	switch wf.Spec.RetryPolicy.Type {
	case cloudv1.RetryPolicyTypeNever:
		return ctrl.Result{}, nil
	case cloudv1.RetryPolicyTypeAlways:
		fallthrough
	default:
		errOcurred := false
		for _, step := range wf.Status.Steps {
			if !step.Done() && step.LastError != "" {
				errOcurred = true
				break
			}
		}
		// requeue immediately if no error occurred
		if !errOcurred {
			return ctrl.Result{Requeue: true}, nil
		}

		// TODO(ideascf): we should get the minimum NextExecuteTime from all failed steps
		// requeue after backoff duration

		wf.Status.NextExecuteTime = &metav1.Time{Time: now.Add(wf.Spec.RetryPolicy.Backoff.Duration)}
		if err := r.client.Status().Update(ctx, wf); err != nil {
			return ctrl.Result{}, err
		}
		return ctrl.Result{RequeueAfter: wf.Spec.RetryPolicy.Backoff.Duration}, nil
	}
}

func (r *WorkflowReconciler) handleWorkflowDone(ctx context.Context, wf *cloudv1.Workflow) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	// remove finalizer
	if controllerutil.RemoveFinalizer(wf, Finalizer) {
		logger.Info("wf is done, remove finalizer", "phase", wf.Status.Phase)
		if err := r.client.Update(ctx, wf); err != nil {
			return ctrl.Result{}, fmt.Errorf("update configChange to remove finalizer: %w", err)
		}
		return ctrl.Result{}, nil
	}

	// GC
	now := r.clock.Now()
	timeToDelete := r.calculateTimeToDelete(wf)
	logger.WithValues("TTL", wf.Spec.TTL.Duration, "now", now, "timeToDelete", timeToDelete)
	if now.After(timeToDelete) {
		logger.Info("Workflow can be deleted")
		return ctrl.Result{}, r.client.Delete(ctx, wf)
	}

	// requeue
	interval := timeToDelete.Sub(now)
	LogAndEmitEvent(ctx, r.eventRecorder, wf, corev1.EventTypeNormal, fmt.Sprintf("The Workflow will be deleted at %s", timeToDelete), nil)
	return ctrl.Result{RequeueAfter: interval}, nil
}

func setWorkflowAsFailed(ctx context.Context, localClient client.Client, wf *cloudv1.Workflow, reason string) error {
	if err := wf.Fail(ctx, reason); err != nil {
		return err
	}
	return localClient.Status().Update(ctx, wf)
}

// nolint:unused
func setWorkflowAsCompleted(ctx context.Context, localClient client.Client, wf *cloudv1.Workflow, reason string) error {
	if err := wf.Complete(ctx, reason); err != nil {
		return err
	}
	return localClient.Status().Update(ctx, wf)
}

func inflateStepInputs(ctx context.Context, wf *cloudv1.Workflow, inputs map[string]interface{}) (map[string]interface{}, error) {
	inflated := map[string]interface{}{}
	for k, v := range inputs {
		val := reflect.ValueOf(v)
		if val.Kind() != reflect.String {
			inflated[k] = v
			continue
		}
		vv := val.String()
		if !strings.HasPrefix(vv, "$") {
			inflated[k] = vv
			continue
		}

		// $steps.{stepName}.outputs.{key}
		// or $wf.spec.inputs.{key}
		split := strings.SplitN(strings.TrimPrefix(vv, "$"), ".", 2)
		if len(split) != 2 {
			return nil, fmt.Errorf("inflateStepInputs: invalid step input variable, key=%s,value=%s", k, vv)
		}
		switch split[0] {
		case "steps":
			value, err := inflateFromSteps(ctx, wf, k, vv)
			if err != nil {
				return nil, fmt.Errorf("inflateFromSteps: %w", err)
			}
			inflated[k] = value
		case "wf":
			value, err := inflateFromWorkflow(ctx, wf, k, vv)
			if err != nil {
				return nil, fmt.Errorf("inflateFromWorkflow: %w", err)
			}
			inflated[k] = value
		default:
			return nil, fmt.Errorf("inflateStepInputs: unknown variable prefix, key=%s,value=%s", k, vv)
		}
	}
	return inflated, nil
}

func inflateFromSteps(ctx context.Context, wf *cloudv1.Workflow, k, vv string) (interface{}, error) {
	if !strings.HasPrefix(vv, "$") {
		return vv, nil
	}

	// $steps.{stepName}.outputs.{key}
	split := strings.Split(strings.TrimPrefix(vv, "$"), ".")
	if len(split) != 4 {
		return nil, fmt.Errorf("inflateFromSteps: invalid step input variable, key=%s,value=%s", k, vv)
	}
	fromCategory, fromStepName, fromWhere, fromKey := split[0], split[1], split[2], split[3]
	if fromCategory != "steps" {
		return nil, fmt.Errorf("inflateFromSteps: fromCategory must be 'steps' but got '%s'", fromCategory)
	}
	switch fromWhere {
	// Get variable value from step's outputs
	case "outputs":
		fromStepStatus, ok := wf.StepStatus(ctx, fromStepName)
		if !ok {
			return nil, fmt.Errorf("inflateFromSteps: fromStep '%s' is not found", fromStepName)
		}
		if fromStepStatus.Phase != cloudv1.WorkflowStepPhaseCompleted {
			return nil, fmt.Errorf("inflateFromSteps: phase(%s) of fromStep '%s' is not Completed", fromStepStatus.Phase, fromStepName)
		}
		value, ok := fromStepStatus.Outputs[fromKey]
		if !ok {
			return nil, fmt.Errorf("inflateFromSteps: key(%s) is not found in outputs of fromStep '%s'", fromKey, fromStepName)
		}
		return value, nil
	// Get variable value from step's status
	case "status":
		fromStepStatus, ok := wf.StepStatus(ctx, fromStepName)
		if !ok {
			return nil, fmt.Errorf("inflateFromSteps: fromStep '%s' is not found", fromStepName)
		}
		if fromStepStatus.Phase != cloudv1.WorkflowStepPhaseCompleted {
			return nil, fmt.Errorf("inflateFromSteps: phase(%s) of fromStep '%s' is not Completed", fromStepStatus.Phase, fromStepName)
		}
		value, err := getValueFromStepStatus(fromStepStatus, fromKey)
		if err != nil {
			return nil, fmt.Errorf("getValueFromStepStatus: %w", err)
		}
		return value, nil
	default:
		return nil, fmt.Errorf("inflateFromSteps: fromWhere must be 'outputs' or 'status but got '%s'", fromWhere)
	}
}

func inflateFromWorkflow(ctx context.Context, wf *cloudv1.Workflow, k, vv string) (interface{}, error) {
	if !strings.HasPrefix(vv, "$") {
		return vv, nil
	}

	// $wf.spec.inputs.{key}
	split := strings.Split(strings.TrimPrefix(vv, "$"), ".")
	if len(split) != 4 {
		return nil, fmt.Errorf("inflateFromWorkflow: invalid workflow input variable, key=%s,value=%s", k, vv)
	}
	fromCategory, fromSubCategory, fromWhere, fromKey := split[0], split[1], split[2], split[3]
	if fromCategory != "wf" {
		return nil, fmt.Errorf("inflateFromWorkflow: fromCategory must be 'wf' but got '%s'", fromCategory)
	}
	if fromSubCategory != "spec" {
		return nil, fmt.Errorf("inflateFromWorkflow: fromSubCategory must be 'spec' but got '%s'", fromSubCategory)
	}
	switch fromWhere {
	case "inputs":
		value, ok := wf.Spec.Inputs[fromKey]
		if !ok {
			return nil, fmt.Errorf("inflateFromWorkflow: key(%s) is not found in inputs of workflow", k)
		}
		return value, nil
	default:
		return nil, fmt.Errorf("inflateFromWorkflow: fromWhere must be 'inputs' but got '%s'", fromWhere)
	}
}

func getValueFromStepStatus(status *cloudv1.WorkflowStepStatus, key string) (string, error) {
	switch key {
	case "beginTime":
		if status.BeginAt == nil {
			return "", nil
		}
		return status.BeginAt.Format(time.RFC3339), nil
	case "endTime":
		if status.EndAt == nil {
			return "", nil
		}
		return status.EndAt.Format(time.RFC3339), nil
	default:
		return "", fmt.Errorf("unsupported key: %s", key)
	}
}

// LogAndEmitEvent logs the event and emits it using the provided logger and eventRecorder.
// The event type determines how the event is logged.
func LogAndEmitEvent(ctx context.Context, eventRecorder record.EventRecorder, obj runtime.Object, eventType, msg string, err error) {
	logger := log.FromContext(ctx)

	switch eventType {
	case corev1.EventTypeNormal:
		logger.Info(msg)
	case corev1.EventTypeWarning:
		logger.Error(err, msg)
	default:
		logger.Info(msg)
	}
	eventRecorder.Event(obj, eventType, "", msg)
}