package integration

import (
	"context"
	"fmt"
	"math/rand"
	"time"

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	"github.com/samber/lo"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/util/retry"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
)

var _ = Describe("Workflow controller", func() {
	var (
		testNamespace string
	)

	BeforeEach(func() {
		testNamespace = genNamespace()
		ns := &corev1.Namespace{
			ObjectMeta: metav1.ObjectMeta{
				Name: testNamespace,
			},
		}
		Expect(k8sClient.Create(context.Background(), ns)).To(Succeed())
	})

	Context("Basic workflow execution", func() {
		It("Should complete a workflow with two echo actions within 10 seconds", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-echo-workflow", []cloudv1.WorkflowStep{
				{
					Name: "echo-step-1",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "hello",
							"value":   "world",
						},
					},
				},
				{
					Name: "echo-step-2",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"result": "$steps.echo-step-1.outputs.message",
						},
					},
				},
			})

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for workflow to complete")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCompleted
			}, "10s", "500ms").Should(BeTrue())

			By("Verifying workflow status")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			Expect(actualWF.Status.Phase).To(Equal(cloudv1.WorkflowPhaseCompleted))
			Expect(actualWF.Status.Progress).To(Equal("2/2"))
			Expect(actualWF.Status.FinishedSteps).To(HaveLen(2))
			Expect(actualWF.Status.RunningSteps).To(BeEmpty())
			Expect(actualWF.Status.EndingSteps).To(BeEmpty())

			By("Verifying step outputs")
			step1Status, ok := actualWF.StepStatus(ctx, "echo-step-1")
			Expect(ok).To(BeTrue())
			Expect(step1Status.Phase).To(Equal(cloudv1.WorkflowStepPhaseCompleted))
			Expect(step1Status.Outputs).To(HaveKeyWithValue("message", "hello"))
			Expect(step1Status.Outputs).To(HaveKeyWithValue("value", "world"))

			step2Status, ok := actualWF.StepStatus(ctx, "echo-step-2")
			Expect(ok).To(BeTrue())
			Expect(step2Status.Phase).To(Equal(cloudv1.WorkflowStepPhaseCompleted))
			Expect(step2Status.Outputs).To(HaveKeyWithValue("result", "hello"))
		}, NodeTimeout(time.Second*15))

		It("Should complete a workflow with sleep action", func(ctx context.Context) {
			sleepDuration := "5s"
			wf := newWorkflow(testNamespace, "test-sleep-workflow", []cloudv1.WorkflowStep{
				{
					Name: "sleep-step",
					Uses: consts.WorkflowActionSleep,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"duration": sleepDuration,
						},
					},
				},
			})
			// reduce the retry backoff for faster test
			wf.Spec.RetryPolicy = &cloudv1.WorkflowRetryPolicy{
				Type:    cloudv1.RetryPolicyTypeAlways,
				Backoff: lo.ToPtr(metav1.Duration{Duration: time.Second}),
			}

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for workflow to complete")
			actualWF := &cloudv1.Workflow{}
			Eventually(func() bool {
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCompleted
			}, "10s", "500ms").Should(BeTrue())

			By("Verifying sleep duration")
			duration := actualWF.Status.EndAt.Time.Sub(actualWF.Status.BeginAt.Time)
			// Sleep should take at least the specified duration
			Expect(duration).To(BeNumerically(">=", lo.Must1(time.ParseDuration(sleepDuration))))

			By("Verifying workflow status")
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			Expect(actualWF.Status.Phase).To(Equal(cloudv1.WorkflowPhaseCompleted))
			Expect(actualWF.Status.Progress).To(Equal("1/1"))
			Expect(actualWF.Status.FinishedSteps).To(HaveLen(1))

			stepStatus, ok := actualWF.StepStatus(ctx, "sleep-step")
			Expect(ok).To(BeTrue())
			Expect(stepStatus.Phase).To(Equal(cloudv1.WorkflowStepPhaseCompleted))
		}, NodeTimeout(time.Second*15))
	})

	Context("Variable passing between steps", func() {
		It("Should pass variables from workflow inputs to steps", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-wf-inputs", []cloudv1.WorkflowStep{
				{
					Name: "echo-from-wf-inputs",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"input-value": "$wf.spec.inputs.test-key",
						},
					},
				},
			})
			wf.Spec.Inputs = map[string]string{
				"test-key": "test-value",
			}

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for workflow to complete")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCompleted
			}, "10s", "500ms").Should(BeTrue())

			By("Verifying step outputs")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			stepStatus, ok := actualWF.StepStatus(ctx, "echo-from-wf-inputs")
			Expect(ok).To(BeTrue())
			Expect(stepStatus.Outputs).To(HaveKeyWithValue("input-value", "test-value"))
		}, NodeTimeout(time.Second*10))

		It("Should chain multiple steps with variable passing", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-chained-steps", []cloudv1.WorkflowStep{
				{
					Name: "step-1",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"data": "initial-data",
						},
					},
				},
				{
					Name: "step-2",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"from-step-1": "$steps.step-1.outputs.data",
						},
					},
				},
				{
					Name: "step-3",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"from-step-2": "$steps.step-2.outputs.from-step-1",
						},
					},
				},
			})

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for workflow to complete")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCompleted
			}, "15s", "500ms").Should(BeTrue())

			By("Verifying all steps completed and variables passed")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			Expect(actualWF.Status.Progress).To(Equal("3/3"))

			step3Status, ok := actualWF.StepStatus(ctx, "step-3")
			Expect(ok).To(BeTrue())
			Expect(step3Status.Phase).To(Equal(cloudv1.WorkflowStepPhaseCompleted))
			Expect(step3Status.Outputs).To(HaveKeyWithValue("from-step-2", "initial-data"))
		}, NodeTimeout(time.Second*20))
	})

	Context("Workflow scheduling", func() {
		It("Should execute workflow at scheduled time", func(ctx context.Context) {
			scheduleTime := time.Now().Add(5 * time.Second)
			wf := newWorkflow(testNamespace, "test-scheduled-workflow", []cloudv1.WorkflowStep{
				{
					Name: "scheduled-echo",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "scheduled",
						},
					},
				},
			})
			wf.Spec.Schedule = &metav1.Time{Time: scheduleTime}

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Verifying workflow stays in Pending phase until schedule time")
			Consistently(func() cloudv1.WorkflowPhase {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return cloudv1.WorkflowPhaseNone
				}
				return actualWF.Status.Phase
			}, "2s", "200ms").Should(BeElementOf([]cloudv1.WorkflowPhase{cloudv1.WorkflowPhaseNone, cloudv1.WorkflowPhasePending}))

			By("Waiting for workflow to complete")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCompleted
			}, "5s", "500ms").Should(BeTrue())
		}, NodeTimeout(time.Second*15))
	})

	Context("Workflow pause and resume", func() {
		It("Should not reconcile when workflow is paused", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-paused-workflow", []cloudv1.WorkflowStep{
				{
					Name: "paused-echo",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "should-not-execute",
						},
					},
				},
			})
			wf.Spec.Paused = true

			By("Creating paused workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Verifying workflow stays uninitialized")
			Consistently(func() cloudv1.WorkflowPhase {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return cloudv1.WorkflowPhaseNone
				}
				return actualWF.Status.Phase
			}, "3s", "500ms").Should(Equal(cloudv1.WorkflowPhaseNone))

			By("Unpausing workflow")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())
			actualWF.Spec.Paused = false
			Expect(k8sClient.Update(ctx, actualWF)).To(Succeed())

			By("Waiting for workflow to complete after unpause")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCompleted
			}, "10s", "500ms").Should(BeTrue())
		}, NodeTimeout(time.Second*15))
	})

	Context("Workflow TTL and garbage collection", func() {
		It("Should delete completed workflow after TTL expires", func(ctx context.Context) {
			ttl := 2 * time.Second
			wf := newWorkflow(testNamespace, "test-ttl-workflow", []cloudv1.WorkflowStep{
				{
					Name: "ttl-echo",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "ttl-test",
						},
					},
				},
			})
			wf.Spec.TTL = &metav1.Duration{Duration: ttl}

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for workflow to complete")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCompleted
			}, "10s", "500ms").Should(BeTrue())

			By("Waiting for workflow to be deleted after TTL")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				return err != nil && client.IgnoreNotFound(err) == nil
			}, "10s", "500ms").Should(BeTrue())
		}, NodeTimeout(time.Second*25))
	})

	Context("Workflow with RetryPolicy", func() {
		It("Should have default retry policy set to Always", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-default-retry", []cloudv1.WorkflowStep{
				{
					Name: "echo-step",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "test",
						},
					},
				},
			})

			By("Creating workflow without explicit retry policy")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for workflow to be initialized with default values")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase != cloudv1.WorkflowPhaseNone
			}, "5s", "500ms").Should(BeTrue())

			By("Verifying default retry policy")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			// Default() should have been called by the controller
			Expect(actualWF.Spec.RetryPolicy).NotTo(BeNil())
			Expect(actualWF.Spec.RetryPolicy.Type).To(Equal(cloudv1.RetryPolicyTypeAlways))

			By("Waiting for workflow to complete")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCompleted
			}, "10s", "500ms").Should(BeTrue())
		}, NodeTimeout(time.Second*15))
	})

	Context("Workflow on error", func() {
		It("Should abort workflow when abort action returns wfAbortedError", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-abort-workflow", []cloudv1.WorkflowStep{
				{
					Name: "echo-before-abort",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "before-abort",
						},
					},
				},
				{
					Name: "abort-step",
					Uses: consts.WorkflowActionAbort,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"abort":   true,
							"message": "Workflow aborted by user request",
						},
					},
				},
				{
					Name: "echo-after-abort",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "should-not-execute",
						},
					},
				},
			})

			By("Creating workflow with abort step")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for workflow to complete with abort")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCompleted
			}, "10s", "500ms").Should(BeTrue())

			By("Verifying workflow was aborted (marked as Completed)")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			Expect(actualWF.Status.Phase).To(Equal(cloudv1.WorkflowPhaseCompleted))
			Expect(actualWF.Status.Reason).To(Equal("Aborted"))

			By("Verifying first step completed successfully")
			step1Status, ok := actualWF.StepStatus(ctx, "echo-before-abort")
			Expect(ok).To(BeTrue())
			Expect(step1Status.Phase).To(Equal(cloudv1.WorkflowStepPhaseCompleted))

			By("Verifying abort step failed with abort reason")
			abortStepStatus, ok := actualWF.StepStatus(ctx, "abort-step")
			Expect(ok).To(BeTrue())
			Expect(abortStepStatus.Phase).To(Equal(cloudv1.WorkflowStepPhaseFailed))
			Expect(abortStepStatus.Reason).To(ContainSubstring("aborted"))

			By("Verifying third step was not executed")
			step3Status, ok := actualWF.StepStatus(ctx, "echo-after-abort")
			Expect(ok).To(BeTrue())
			Expect(step3Status.Phase).To(Equal(cloudv1.WorkflowStepPhasePending))
		}, NodeTimeout(time.Second*15))

		It("Should handle abort in the middle of workflow with proper progress", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-abort-mid-workflow", []cloudv1.WorkflowStep{
				{
					Name: "step-1",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "step-1",
						},
					},
				},
				{
					Name: "step-2",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "step-2",
						},
					},
				},
				{
					Name: "abort-step",
					Uses: consts.WorkflowActionAbort,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"abort":   true,
							"message": "Abort at step 3",
						},
					},
				},
				{
					Name: "step-4",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "step-4-not-executed",
						},
					},
				},
				{
					Name: "step-5",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "step-5-not-executed",
						},
					},
				},
			})

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for workflow to be aborted")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCompleted &&
					actualWF.Status.Reason == "Aborted"
			}, "10s", "500ms").Should(BeTrue())

			By("Verifying progress reflects completed steps before abort")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			// Progress should show 2 completed steps (step-1 and step-2)
			// The abort step is failed, not completed
			Expect(lo.CountBy(actualWF.Status.Steps, func(s *cloudv1.WorkflowStepStatus) bool {
				return s.Phase == cloudv1.WorkflowStepPhaseCompleted
			})).To(Equal(2))

			Expect(lo.CountBy(actualWF.Status.Steps, func(s *cloudv1.WorkflowStepStatus) bool {
				return s.Phase == cloudv1.WorkflowStepPhasePending
			})).To(Equal(2))

			Expect(lo.CountBy(actualWF.Status.Steps, func(s *cloudv1.WorkflowStepStatus) bool {
				return s.Phase == cloudv1.WorkflowStepPhaseFailed
			})).To(Equal(1))
		}, NodeTimeout(time.Second*15))
	})

	Context("Workflow cancellation", func() {
		It("Should cancel a pending workflow", func(ctx context.Context) {
			// Use schedule to keep workflow in Pending phase
			scheduleTime := time.Now().Add(10 * time.Second)
			wf := newWorkflow(testNamespace, "test-cancel-pending", []cloudv1.WorkflowStep{
				{
					Name: "echo-step",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "should-not-execute",
						},
					},
				},
			})
			wf.Spec.Schedule = &metav1.Time{Time: scheduleTime}

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for workflow to be in Pending phase")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhasePending
			}, "5s", "500ms").Should(BeTrue())

			By("Adding cancel annotation")
			Expect(updateWorkflowWithRetry(ctx, k8sClient, wf, func(wf *cloudv1.Workflow) {
				if wf.Annotations == nil {
					wf.Annotations = make(map[string]string)
				}
				wf.Annotations[consts.AnnoCancelWorkflow] = "true"
			})).To(Succeed())

			By("Waiting for workflow to be cancelled")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCancelled
			}, "10s", "500ms").Should(BeTrue())

			By("Verifying workflow status")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			Expect(actualWF.Status.Phase).To(Equal(cloudv1.WorkflowPhaseCancelled))
			Expect(actualWF.Status.FinishedSteps).To(HaveLen(1))

			By("Verifying step is cancelled")
			stepStatus, ok := actualWF.StepStatus(ctx, "echo-step")
			Expect(ok).To(BeTrue())
			Expect(stepStatus.Phase).To(Equal(cloudv1.WorkflowStepPhaseCancelled))
		}, NodeTimeout(time.Second*20))

		It("Should cancel a running workflow with running step", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-cancel-running", []cloudv1.WorkflowStep{
				{
					Name: "sleep-step",
					Uses: consts.WorkflowActionCancelableSleep,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"duration": "30s",
						},
					},
				},
			})

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for workflow to start running")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				// Workflow should be Running, and step should be in Running or Ending phase
				if actualWF.Status.Phase != cloudv1.WorkflowPhaseRunning {
					return false
				}
				stepStatus, ok := actualWF.StepStatus(ctx, "sleep-step")
				if !ok {
					return false
				}
				// Step can be in Running or Ending phase (Ending happens quickly for cancelable-sleep)
				return stepStatus.Phase == cloudv1.WorkflowStepPhaseRunning ||
					stepStatus.Phase == cloudv1.WorkflowStepPhaseEnding
			}, "5s", "500ms").Should(BeTrue())

			By("Adding cancel annotation")
			Expect(updateWorkflowWithRetry(ctx, k8sClient, wf, func(wf *cloudv1.Workflow) {
				if wf.Annotations == nil {
					wf.Annotations = make(map[string]string)
				}
				wf.Annotations[consts.AnnoCancelWorkflow] = "true"
			})).To(Succeed())

			By("Waiting for workflow to be cancelled")
			// Note: handleCancelling may wait for NextExecuteTime (up to sleep duration),
			// so we need a longer timeout to account for this delay
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCancelled
			}, "40s", "1s").Should(BeTrue())

			By("Verifying workflow status")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			Expect(actualWF.Status.Phase).To(Equal(cloudv1.WorkflowPhaseCancelled))
			Expect(actualWF.Status.FinishedSteps).To(HaveLen(1))
			Expect(actualWF.Status.RunningSteps).To(BeEmpty())
			Expect(actualWF.Status.EndingSteps).To(BeEmpty())
			Expect(actualWF.Status.CancellingSteps).To(BeEmpty())
			Expect(actualWF.Status.CancelEndingSteps).To(BeEmpty())

			By("Verifying step transitioned through cancellation phases")
			stepStatus, ok := actualWF.StepStatus(ctx, "sleep-step")
			Expect(ok).To(BeTrue())
			Expect(stepStatus.Phase).To(Equal(cloudv1.WorkflowStepPhaseCancelled))
		}, NodeTimeout(time.Second*60))

		It("Should cancel a workflow with multiple steps and step groups", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-cancel-multi-steps", []cloudv1.WorkflowStep{
				{
					Name:      "step-1",
					StepGroup: "group1",
					Uses:      consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "step-1",
						},
					},
				},
				{
					Name:      "step-2",
					StepGroup: "group1",
					Uses:      consts.WorkflowActionCancelableSleep,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"duration": "30s",
						},
					},
				},
				{
					Name:      "step-3",
					StepGroup: "group2",
					Uses:      consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "step-3",
						},
					},
				},
			})

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for step-2 to start running")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				stepStatus, ok := actualWF.StepStatus(ctx, "step-2")
				if !ok {
					return false
				}
				// Step can be in Running or Ending phase (Ending happens quickly for cancelable-sleep)
				return stepStatus.Phase == cloudv1.WorkflowStepPhaseRunning ||
					stepStatus.Phase == cloudv1.WorkflowStepPhaseEnding
			}, "10s", "500ms").Should(BeTrue())

			By("Adding cancel annotation")
			Expect(updateWorkflowWithRetry(ctx, k8sClient, wf, func(wf *cloudv1.Workflow) {
				if wf.Annotations == nil {
					wf.Annotations = make(map[string]string)
				}
				wf.Annotations[consts.AnnoCancelWorkflow] = "true"
			})).To(Succeed())

			By("Waiting for workflow to be cancelled")
			// Note: handleCancelling may wait for NextExecuteTime (up to sleep duration),
			// so we need a longer timeout to account for this delay
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCancelled
			}, "40s", "1s").Should(BeTrue())

			By("Verifying workflow status")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			Expect(actualWF.Status.Phase).To(Equal(cloudv1.WorkflowPhaseCancelled))

			By("Verifying step-1 is completed (was already done)")
			step1Status, ok := actualWF.StepStatus(ctx, "step-1")
			Expect(ok).To(BeTrue())
			Expect(step1Status.Phase).To(Equal(cloudv1.WorkflowStepPhaseCompleted))

			// step-2 should be completed because the step-1 in the same group was completed
			By("Verifying step-2 is completed (was running)")
			step2Status, ok := actualWF.StepStatus(ctx, "step-2")
			Expect(ok).To(BeTrue())
			Expect(step2Status.Phase).To(Equal(cloudv1.WorkflowStepPhaseCompleted))

			By("Verifying step-3 is cancelled (was pending in inactive group)")
			step3Status, ok := actualWF.StepStatus(ctx, "step-3")
			Expect(ok).To(BeTrue())
			Expect(step3Status.Phase).To(Equal(cloudv1.WorkflowStepPhaseCancelled))
		}, NodeTimeout(time.Second*60))

		It("Should handle cancelling workflow with ending step", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-cancel-ending", []cloudv1.WorkflowStep{
				{
					Name: "echo-step",
					Uses: consts.WorkflowActionEcho,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "echo",
						},
					},
				},
				{
					Name: "sleep-step",
					Uses: consts.WorkflowActionCancelableSleep,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"duration": "30s",
						},
					},
				},
			})

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for sleep-step to be in Ending phase")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				stepStatus, ok := actualWF.StepStatus(ctx, "sleep-step")
				return ok && stepStatus.Phase == cloudv1.WorkflowStepPhaseEnding
			}, "10s", "500ms").Should(BeTrue())

			By("Adding cancel annotation")
			Expect(updateWorkflowWithRetry(ctx, k8sClient, wf, func(wf *cloudv1.Workflow) {
				if wf.Annotations == nil {
					wf.Annotations = make(map[string]string)
				}
				wf.Annotations[consts.AnnoCancelWorkflow] = "true"
			})).To(Succeed())

			By("Waiting for workflow to be cancelled")
			// Note: handleCancelling may wait for NextExecuteTime (up to sleep duration),
			// so we need a longer timeout to account for this delay
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCancelled
			}, "40s", "1s").Should(BeTrue())

			By("Verifying workflow status")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			Expect(actualWF.Status.Phase).To(Equal(cloudv1.WorkflowPhaseCancelled))

			By("Verifying sleep-step is cancelled (was ending)")
			stepStatus, ok := actualWF.StepStatus(ctx, "sleep-step")
			Expect(ok).To(BeTrue())
			Expect(stepStatus.Phase).To(Equal(cloudv1.WorkflowStepPhaseCancelled))
		}, NodeTimeout(time.Second*60))

		It("Should handle cancel annotation on already cancelling workflow", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-cancel-already-cancelling", []cloudv1.WorkflowStep{
				{
					Name: "sleep-step",
					Uses: consts.WorkflowActionCancelableSleep,
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"duration": "30s",
						},
					},
				},
			})

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for workflow to start running")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseRunning
			}, "5s", "500ms").Should(BeTrue())

			By("Adding cancel annotation first time")
			Expect(updateWorkflowWithRetry(ctx, k8sClient, wf, func(wf *cloudv1.Workflow) {
				if wf.Annotations == nil {
					wf.Annotations = make(map[string]string)
				}
				wf.Annotations[consts.AnnoCancelWorkflow] = "true"
			})).To(Succeed())

			By("Waiting for workflow to enter Cancelling phase")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCancelling
			}, "5s", "500ms").Should(BeTrue())

			By("Adding cancel annotation again (should be ignored)")
			Expect(updateWorkflowWithRetry(ctx, k8sClient, wf, func(wf *cloudv1.Workflow) {
				if wf.Annotations == nil {
					wf.Annotations = make(map[string]string)
				}
				wf.Annotations[consts.AnnoCancelWorkflow] = "true"
			})).To(Succeed())

			By("Waiting for workflow to be cancelled")
			// Note: handleCancelling may wait for NextExecuteTime (up to sleep duration),
			// so we need a longer timeout to account for this delay
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCancelled
			}, "40s", "1s").Should(BeTrue())

			By("Verifying workflow was cancelled successfully")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			Expect(actualWF.Status.Phase).To(Equal(cloudv1.WorkflowPhaseCancelled))
		}, NodeTimeout(time.Second*60))

		It("Should cancel a workflow with non-cancelable step (step continues execution)", func(ctx context.Context) {
			wf := newWorkflow(testNamespace, "test-cancel-non-cancelable", []cloudv1.WorkflowStep{
				{
					Name:      "echo-step",
					Uses:      consts.WorkflowActionEcho,
					StepGroup: "group1",
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "echo",
						},
					},
				},
				{
					Name:      "sleep-step",
					Uses:      consts.WorkflowActionSleep, // Non-cancelable action
					StepGroup: "group2",
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"duration": "5s",
						},
					},
				},
				{
					Name:      "pending-step",
					Uses:      consts.WorkflowActionEcho,
					StepGroup: "group3",
					Inputs: cloudv1.InputsWrapper{
						Object: map[string]interface{}{
							"message": "should-not-execute",
						},
					},
				},
			})
			// reduce the retry backoff for faster test
			wf.Spec.RetryPolicy = &cloudv1.WorkflowRetryPolicy{
				Type:    cloudv1.RetryPolicyTypeAlways,
				Backoff: lo.ToPtr(metav1.Duration{Duration: time.Second}),
			}

			By("Creating workflow")
			Expect(k8sClient.Create(ctx, wf)).To(Succeed())

			By("Waiting for sleep-step to start running")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				// Workflow should be Running, and step should be in Running or Ending phase
				if actualWF.Status.Phase != cloudv1.WorkflowPhaseRunning {
					return false
				}
				stepStatus, ok := actualWF.StepStatus(ctx, "sleep-step")
				if !ok {
					return false
				}
				// Step can be in Running or Ending phase (Ending happens quickly for sleep)
				return stepStatus.Phase == cloudv1.WorkflowStepPhaseRunning ||
					stepStatus.Phase == cloudv1.WorkflowStepPhaseEnding
			}, "5s", "500ms").Should(BeTrue())

			By("Adding cancel annotation")
			Expect(updateWorkflowWithRetry(ctx, k8sClient, wf, func(wf *cloudv1.Workflow) {
				if wf.Annotations == nil {
					wf.Annotations = make(map[string]string)
				}
				wf.Annotations[consts.AnnoCancelWorkflow] = "true"
			})).To(Succeed())

			By("Waiting for workflow to enter Cancelling phase")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCancelling
			}, "5s", "500ms").Should(BeTrue())

			By("Waiting for sleep-step to complete (non-cancelable step continues execution)")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				stepStatus, ok := actualWF.StepStatus(ctx, "sleep-step")
				return ok && stepStatus.Phase == cloudv1.WorkflowStepPhaseCompleted
			}, "10s", "500ms").Should(BeTrue())

			By("Waiting for workflow to be cancelled")
			Eventually(func() bool {
				actualWF := &cloudv1.Workflow{}
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: wf.Namespace,
					Name:      wf.Name,
				}, actualWF)
				if err != nil {
					return false
				}
				return actualWF.Status.Phase == cloudv1.WorkflowPhaseCancelled
			}, "10s", "500ms").Should(BeTrue())

			By("Verifying workflow status")
			actualWF := &cloudv1.Workflow{}
			Expect(k8sClient.Get(ctx, types.NamespacedName{
				Namespace: wf.Namespace,
				Name:      wf.Name,
			}, actualWF)).To(Succeed())

			Expect(actualWF.Status.Phase).To(Equal(cloudv1.WorkflowPhaseCancelled))

			By("Verifying echo-step is completed (was already done)")
			echoStepStatus, ok := actualWF.StepStatus(ctx, "echo-step")
			Expect(ok).To(BeTrue())
			Expect(echoStepStatus.Phase).To(Equal(cloudv1.WorkflowStepPhaseCompleted))

			By("Verifying sleep-step is completed (non-cancelable step continued execution)")
			sleepStepStatus, ok := actualWF.StepStatus(ctx, "sleep-step")
			Expect(ok).To(BeTrue())
			Expect(sleepStepStatus.Phase).To(Equal(cloudv1.WorkflowStepPhaseCompleted))

			By("Verifying pending-step is cancelled (was pending)")
			pendingStepStatus, ok := actualWF.StepStatus(ctx, "pending-step")
			Expect(ok).To(BeTrue())
			Expect(pendingStepStatus.Phase).To(Equal(cloudv1.WorkflowStepPhaseCancelled))
		}, NodeTimeout(time.Second*20))
	})

})

// Helper functions

// updateWorkflowWithRetry updates a workflow with retry logic to handle resource version conflicts
func updateWorkflowWithRetry(ctx context.Context, c client.Client, wf *cloudv1.Workflow, updateFn func(*cloudv1.Workflow)) error {
	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
		// Re-fetch the workflow to get the latest version
		latestWF := &cloudv1.Workflow{}
		if err := c.Get(ctx, types.NamespacedName{
			Namespace: wf.Namespace,
			Name:      wf.Name,
		}, latestWF); err != nil {
			return err
		}
		// Apply the update function
		updateFn(latestWF)
		// Try to update
		if err := c.Update(ctx, latestWF); err != nil {
			return err
		}
		// Update the original object reference
		*wf = *latestWF
		return nil
	})
}

func newWorkflow(namespace, name string, steps []cloudv1.WorkflowStep) *cloudv1.Workflow {
	return &cloudv1.Workflow{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-%d", name, rand.Int()),
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ReconcilerLabelKey: workflowReconcilerLabel,
			},
		},
		Spec: cloudv1.WorkflowSpec{
			Type:  "Test",
			Steps: steps,
		},
	}
}