package worflow_helper

import (
	"context"
	"fmt"
	"time"

	"github.com/avast/retry-go/v4"
	"github.com/elliotchance/pie/v2"
	"github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/mod"
	"github.com/tidbcloud/dataflow-service-ng/app/workflow"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"go.uber.org/zap"
)

type DagCanceler interface {
	SendCancellations(ctx context.Context, tags map[string]string) error
	WaitDone(ctx context.Context, tags map[string]string) error
}

type DagCancelerImpl struct {
	workflowModel workflow.Workflow
}

func NewDagCanceler(
	workflowModel workflow.Workflow,
) DagCanceler {
	return &DagCancelerImpl{
		workflowModel: workflowModel,
	}
}

func (a *DagCancelerImpl) getToBeCancelDagIns(
	ctx context.Context, tags map[string]string,
) ([]*entity.DagInstance, error) {
	businessID, ok := tags[workflow.BusinessID]
	if !ok {
		return a.listRunningDagInstances(tags)
	}

	// performance issue, Tags queries should use tags with high differences as much as possible.
	// If tags with poor differences are used, query performance will deteriorate
	dagIns, err := a.listRunningDagInstances(map[string]string{workflow.BusinessID: businessID})
	if err != nil {
		return nil, err
	}

	localTagFilter := make(map[string]string)
	for k, v := range tags {
		if k != workflow.BusinessID {
			localTagFilter[k] = v
		}
	}
	var ret []*entity.DagInstance
	for i := range dagIns {
		d := dagIns[i]
		if result := a.filterByLocalTagFilter(d, localTagFilter); !result {
			log.GetLogger(ctx).Info("filter by local tags", zap.String("dag_ins_id", d.ID))
			continue
		}
		ret = append(ret, d)
	}
	return ret, nil
}

func (a *DagCancelerImpl) listRunningDagInstances(tags map[string]string) ([]*entity.DagInstance, error) {
	listReq := &mod.ListDagInstanceInput{
		Status: []entity.DagInstanceStatus{
			entity.DagInstanceStatusInit, entity.DagInstanceStatusScheduled, entity.DagInstanceStatusRunning,
		},
		Tags: tags,
	}
	dagIns, err := a.workflowModel.ListDagIns(listReq)
	return dagIns, err
}

//nolint:gomnd
func (a *DagCancelerImpl) SendCancellations(
	ctx context.Context, tags map[string]string,
) error {
	logger := log.GetBaseLogger(ctx)
	retryErr := retry.Do(func() error {
		dagIns, listErr := a.getToBeCancelDagIns(ctx, tags)
		if listErr != nil {
			logger.Error("list dag instance error", zap.Error(listErr))
			return listErr
		}
		if len(dagIns) == 0 {
			logger.Info("there is no other dag instance running")
			return nil
		}
		taskInstance := a.getTaskIns(ctx)
		for _, d := range dagIns {
			if taskInstance != nil && d.ID == taskInstance.DagInsID {
				continue
			}
			logger.Info("stop dag instance", zap.Any("dag_instance_id", d.ID))
			stopErr := a.workflowModel.CancelDagIns(d.ID)
			if stopErr != nil {
				logger.Warn("stop dag instance error", zap.Error(stopErr))
				return stopErr
			}
		}
		return nil
	},
		retry.Context(ctx),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(time.Second*1),
		retry.Attempts(3),
		retry.OnRetry(func(n uint, err error) {
			logger.Info("action run error", zap.Uint("retry_times", n))
		}),
		retry.LastErrorOnly(true),
	)
	if retryErr != nil {
		logger.Error("stop dag instance error", zap.Error(retryErr))
		return retryErr
	}
	return nil
}

func (a *DagCancelerImpl) filterByLocalTagFilter(
	d *entity.DagInstance, localTagFilter map[string]string,
) bool {
	instanceTags := make(map[string]string)
	for _, tag := range d.Tags {
		instanceTags[tag.Key] = tag.Value
	}
	for k, v := range localTagFilter {
		val, ok := instanceTags[k]
		if !ok {
			return false
		}
		if val != v {
			return false
		}
	}
	return true
}

//nolint:gomnd
func (a *DagCancelerImpl) WaitDone(
	ctx context.Context, tags map[string]string,
) error {
	logger := log.GetBaseLogger(ctx)
	err := retry.Do(func() error {
		dagIns, listErr := a.getToBeCancelDagIns(ctx, tags)
		if listErr != nil {
			logger.Error("list dag instance error", zap.Error(listErr))
			return listErr
		}
		taskInstance := a.getTaskIns(ctx)
		dagIns = pie.Of(dagIns).Filter(func(d *entity.DagInstance) bool {
			if taskInstance == nil {
				return true
			}
			return d.ID != taskInstance.DagInsID
		}).Result
		if len(dagIns) > 0 {
			logger.Warn("there are still dag instance running")
			return fmt.Errorf("there are still %d dag instance running", len(dagIns))
		}
		return nil
	},
		retry.Context(ctx),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(time.Second*15),
		retry.Attempts(4),
		retry.OnRetry(func(n uint, err error) {
			logger.Info("action runAfter error", zap.Uint("retry_times", n))
		}),
		retry.LastErrorOnly(true),
	)
	if err != nil {
		logger.Error("action run after error", zap.Error(err))
		return err
	}
	logger.Info("action run after finished")
	return nil
}

func (a *DagCancelerImpl) getTaskIns(ctx context.Context) *entity.TaskInstance {
	taskIns, ok := entity.CtxRunningTaskIns(ctx)
	if !ok {
		log.GetLogger(ctx).Info("query dag ins is null")
		return nil
	}
	return taskIns
}