package changefeed

import (
	"testing"

	"github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	appworkflow "github.com/tidbcloud/serverless-service/internal/workflow"
)

func TestCreateChangefeedDagRegistersClusterJob(t *testing.T) {
	dag, actions := GetCreateChangefeedDag(nil)

	if !hasAction(actions, appworkflow.ActionRegisterClusterJob) {
		t.Fatalf("create dag actions should include %s", appworkflow.ActionRegisterClusterJob)
	}
	if !hasTask(dag.Tasks, appworkflow.ActionRegisterClusterJob) {
		t.Fatalf("create dag tasks should include %s", appworkflow.ActionRegisterClusterJob)
	}

	register := findTask(dag.Tasks, appworkflow.ActionRegisterClusterJob)
	if len(register.DependOn) != 0 {
		t.Fatalf("%s should be the first task, got dependencies %v",
			appworkflow.ActionRegisterClusterJob, register.DependOn)
	}

	create := findTask(dag.Tasks, appworkflow.ActionCreateOrScaleNewArchCluster)
	if create == nil {
		t.Fatalf("create dag should include %s", appworkflow.ActionCreateOrScaleNewArchCluster)
	}
	if len(create.DependOn) != 1 || create.DependOn[0] != appworkflow.ActionRegisterClusterJob {
		t.Fatalf("%s should depend on %s, got %v",
			appworkflow.ActionCreateOrScaleNewArchCluster, appworkflow.ActionRegisterClusterJob, create.DependOn)
	}
}

func TestDeleteChangefeedDagDeregistersClusterJob(t *testing.T) {
	dag, actions := GetDeleteChangefeedDag(nil, nil)

	if !hasAction(actions, appworkflow.ActionDeregisterClusterJob) {
		t.Fatalf("delete dag actions should include %s", appworkflow.ActionDeregisterClusterJob)
	}
	if !hasTask(dag.Tasks, appworkflow.ActionDeregisterClusterJob) {
		t.Fatalf("delete dag tasks should include %s", appworkflow.ActionDeregisterClusterJob)
	}

	deregister := findTask(dag.Tasks, appworkflow.ActionDeregisterClusterJob)
	if len(deregister.DependOn) != 1 || deregister.DependOn[0] != appworkflow.ActionDeleteNewArchDeregister {
		t.Fatalf("%s should depend on %s, got %v",
			appworkflow.ActionDeregisterClusterJob, appworkflow.ActionDeleteNewArchDeregister, deregister.DependOn)
	}

	final := findTask(dag.Tasks, appworkflow.ActionDeleteNewArchFinal)
	if final == nil {
		t.Fatalf("delete dag should include %s", appworkflow.ActionDeleteNewArchFinal)
	}
	if len(final.DependOn) != 1 || final.DependOn[0] != appworkflow.ActionDeregisterClusterJob {
		t.Fatalf("%s should depend on %s, got %v",
			appworkflow.ActionDeleteNewArchFinal, appworkflow.ActionDeregisterClusterJob, final.DependOn)
	}
}

func hasAction(actions []run.Action, name string) bool {
	for _, action := range actions {
		if action.Name() == name {
			return true
		}
	}
	return false
}

func hasTask(tasks []entity.Task, id string) bool {
	return findTask(tasks, id) != nil
}

func findTask(tasks []entity.Task, id string) *entity.Task {
	for i := range tasks {
		if tasks[i].ID == id {
			return &tasks[i]
		}
	}
	return nil
}