package workflow

import (
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/fastflow/pkg/entity"
)

func TestIsFullStuckIgnoresTimeoutOfPendingTask(t *testing.T) {
	nowSec := time.Now().Unix()
	staleAt := nowSec - int64((3 * executorTimeout).Seconds())
	dag := &entity.DagInstance{BaseInfo: entity.BaseInfo{UpdatedAt: staleAt}}
	tasks := []*entity.TaskInstance{
		{
			BaseInfo: entity.BaseInfo{UpdatedAt: staleAt},
			Status:   entity.TaskInstanceStatusRunning,
		},
		{
			BaseInfo:    entity.BaseInfo{UpdatedAt: staleAt},
			Status:      entity.TaskInstanceStatusInit,
			TimeoutSecs: int((15 * time.Minute).Seconds()),
		},
	}

	require.True(t, isFullStuck(dag, tasks, nowSec))
}

func TestCollectRetryCommandTargetIDsSkipsTerminalTasks(t *testing.T) {
	tests := []struct {
		name     string
		terminal bool
		want     []string
	}{
		{
			name:     "retryable failed and canceled tasks",
			terminal: false,
			want:     []string{"failed", "canceled"},
		},
		{
			name:     "terminal failed and canceled tasks",
			terminal: true,
			want:     []string{},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tasks := []*entity.TaskInstance{
				{
					BaseInfo: entity.BaseInfo{ID: "failed"},
					Status:   entity.TaskInstanceStatusFailed,
				},
				{
					BaseInfo: entity.BaseInfo{ID: "canceled"},
					Status:   entity.TaskInstanceStatusCanceled,
				},
			}
			if tt.terminal {
				tasks[0].Reason = terminalErrorMarker + " failed"
				tasks[1].Reason = terminalErrorMarker + " canceled"
			}

			got := collectRetryCommandTargetIDs([]string{"failed", "canceled"}, tasks)
			require.Equal(t, tt.want, got)
		})
	}
}