package changefeed

import (
	"fmt"

	"github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	appworkflow "github.com/tidbcloud/serverless-service/internal/workflow"
	workflowchangefeed "github.com/tidbcloud/serverless-service/internal/workflow/action/changefeed"
	deleteaction "github.com/tidbcloud/serverless-service/internal/workflow/action/changefeed/delete"
)

func GetDeleteChangefeedDag(
	workflow appworkflow.Workflow,
	hook appworkflow.CDCNewArchHook,
) (*entity.Dag, []run.Action) {
	scheduler := workflowchangefeed.NewSharedScheduler(hook)
	actions := []run.Action{
		deleteaction.NewCancelUnfinishedDagInstancesAction(workflow),
		deleteaction.NewDeleteChangefeedAction(hook),
		deleteaction.NewDeleteOrScaleClusterAction(scheduler),
		deleteaction.NewDeregisterChangefeedAction(hook),
		deleteaction.NewDeregisterClusterJobAction(hook),
		deleteaction.NewDeleteChangefeedFinalAction(hook),
	}

	dag := &entity.Dag{
		BaseInfo: entity.BaseInfo{
			ID: appworkflow.DagDeleteNewArchChangefeed,
		},
		Name: appworkflow.DagDeleteNewArchChangefeed,
		Vars: entity.DagVars{
			"cluster_id":    {},
			"changefeed_id": {},
		},
		Status: entity.DagStatusNormal,
		Tasks: []entity.Task{
			{
				ID:         appworkflow.ActionCancelNewArchDagIns,
				ActionName: appworkflow.ActionCancelNewArchDagIns,
				Params: map[string]any{
					"cluster_id":    fmt.Sprintf("{{%s}}", "cluster_id"),
					"changefeed_id": fmt.Sprintf("{{%s}}", "changefeed_id"),
				},
				TimeoutSecs: 5 * 60,
			},
			{
				ID:         appworkflow.ActionDeleteNewArchFeed,
				ActionName: appworkflow.ActionDeleteNewArchFeed,
				Params: map[string]any{
					"cluster_id":    fmt.Sprintf("{{%s}}", "cluster_id"),
					"changefeed_id": fmt.Sprintf("{{%s}}", "changefeed_id"),
				},
				TimeoutSecs: 10 * 60,
				DependOn:    []string{appworkflow.ActionCancelNewArchDagIns},
			},
			{
				ID:         appworkflow.ActionDeleteOrScaleCluster,
				ActionName: appworkflow.ActionDeleteOrScaleCluster,
				Params: map[string]any{
					"cluster_id":    fmt.Sprintf("{{%s}}", "cluster_id"),
					"changefeed_id": fmt.Sprintf("{{%s}}", "changefeed_id"),
				},
				TimeoutSecs: 30 * 60,
				DependOn:    []string{appworkflow.ActionDeleteNewArchFeed},
			},
			{
				ID:         appworkflow.ActionDeleteNewArchDeregister,
				ActionName: appworkflow.ActionDeleteNewArchDeregister,
				Params: map[string]any{
					"cluster_id":    fmt.Sprintf("{{%s}}", "cluster_id"),
					"changefeed_id": fmt.Sprintf("{{%s}}", "changefeed_id"),
				},
				TimeoutSecs: 3 * 60,
				DependOn:    []string{appworkflow.ActionDeleteOrScaleCluster},
			},
			{
				ID:         appworkflow.ActionDeregisterClusterJob,
				ActionName: appworkflow.ActionDeregisterClusterJob,
				Params: map[string]any{
					"cluster_id":    fmt.Sprintf("{{%s}}", "cluster_id"),
					"changefeed_id": fmt.Sprintf("{{%s}}", "changefeed_id"),
				},
				TimeoutSecs: 3 * 60,
				DependOn:    []string{appworkflow.ActionDeleteNewArchDeregister},
			},
			{
				ID:         appworkflow.ActionDeleteNewArchFinal,
				ActionName: appworkflow.ActionDeleteNewArchFinal,
				Params: map[string]any{
					"cluster_id":    fmt.Sprintf("{{%s}}", "cluster_id"),
					"changefeed_id": fmt.Sprintf("{{%s}}", "changefeed_id"),
				},
				TimeoutSecs: 3 * 60,
				DependOn:    []string{appworkflow.ActionDeregisterClusterJob},
			},
		},
	}
	return dag, actions
}