package del

import (
	"context"
	"errors"

	ffentity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/mod"
	appworkflow "github.com/tidbcloud/serverless-service/internal/workflow"
	"github.com/tidbcloud/serverless-service/internal/workflow/action"
	workflowchangefeed "github.com/tidbcloud/serverless-service/internal/workflow/action/changefeed"
	"go.uber.org/zap"
)

type ActionCancelUnfinishedDagInstances struct {
	workflow appworkflow.Workflow
}

func NewCancelUnfinishedDagInstancesAction(workflow appworkflow.Workflow) *action.Wrapper {
	return action.MustNewActionWrapper(
		appworkflow.ActionCancelNewArchDagIns,
		&ActionCancelUnfinishedDagInstances{workflow: workflow},
	)
}

func (a *ActionCancelUnfinishedDagInstances) Params() any {
	return &workflowchangefeed.CommonParams{}
}

func (a *ActionCancelUnfinishedDagInstances) Run(ctx context.Context, logger *zap.Logger, params any) error {
	input := params.(*workflowchangefeed.CommonParams)
	dagInsList, err := a.listUnfinishedDagIns(input)
	if err != nil {
		return err
	}
	for _, dagIns := range dagInsList {
		logger.Info("cancel unfinished create dag instance", zap.String("dag_ins_id", dagIns.ID))
		if err := a.workflow.CancelDagIns(dagIns.ID); err != nil {
			return err
		}
	}
	return nil
}

func (a *ActionCancelUnfinishedDagInstances) RunAfter(ctx context.Context, logger *zap.Logger, params any) error {
	input := params.(*workflowchangefeed.CommonParams)
	dagInsList, err := a.listUnfinishedDagIns(input)
	if err != nil {
		return err
	}
	if len(dagInsList) > 0 {
		logger.Info("waiting unfinished changefeed dag instances to exit", zap.Int("remaining", len(dagInsList)))
		return errors.New("unfinished changefeed dag instances still running")
	}
	return nil
}

func (a *ActionCancelUnfinishedDagInstances) listUnfinishedDagIns(
	input *workflowchangefeed.CommonParams,
) ([]*ffentity.DagInstance, error) {
	var dagInsList []*ffentity.DagInstance
	for _, dagID := range []string{
		appworkflow.DagCreateNewArchChangefeed,
		appworkflow.DagScaleNewArchCluster,
	} {
		items, err := a.workflow.ListDagIns(&mod.ListDagInstanceInput{
			DagID: dagID,
			Tags: map[string]string{
				appworkflow.TagClusterID:    input.ClusterID,
				appworkflow.TagChangefeedID: input.ChangefeedID,
			},
			Status: []ffentity.DagInstanceStatus{
				ffentity.DagInstanceStatusInit,
				ffentity.DagInstanceStatusScheduled,
				ffentity.DagInstanceStatusRunning,
				ffentity.DagInstanceStatusBlocked,
			},
		})
		if err != nil {
			return nil, err
		}
		dagInsList = append(dagInsList, items...)
	}
	return dagInsList, nil
}