package executorext

import (
	"context"
	"fmt"
	"strings"

	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
)

type DagInstanceFinder interface {
	GetRunningDagInstancesWithFilterFunc(ctx context.Context, plan []string, fn FilterFunc, filters ...DagInsFilter) (executor.ExecTaskIdStatus, error)
	GetRunningDagInstances(ctx context.Context, plan []string, filters ...DagInsFilter) (executor.ExecTaskIdStatus, error)
	GetRunningDagInstance(ctx context.Context, plan string, filters ...DagInsFilter) (*executor.ExecTaskStatusEx, error)
}

type DagInsFilter struct {
	ArgKey    string
	ArgValues []string
}

// FilterFunc converts a list of DagInsFilter into a MatchFilter for server-side filtering.
type FilterFunc func(filterList []DagInsFilter) executor.MatchFilter

type dagInstanceFinder struct {
	dagExecutor executor.RemoteExecutor
}

func NewDagInstanceFinder(dagExecutor executor.RemoteExecutor) DagInstanceFinder {
	return &dagInstanceFinder{
		dagExecutor: dagExecutor,
	}
}

func (t *dagInstanceFinder) GetRunningDagInstancesWithFilterFunc(ctx context.Context, plans []string, filterFn FilterFunc, filters ...DagInsFilter) (executor.ExecTaskIdStatus, error) {

	err := validateFilters(filters)
	if err != nil {
		return nil, err
	}

	res := executor.NewExecTaskIdStatus()
	for _, planName := range plans {
		dagIns, err := t.dagExecutor.QueryRunnings(ctx, planName, filterFn(filters))
		if err != nil {
			return nil, fmt.Errorf("failed to QueryRunnings for plan %s, error: %w", planName, err)
		}
		for execId, ins := range dagIns {
			res[execId] = ins
		}
	}

	return res, nil
}

var DagInsNotFoundErr = fmt.Errorf("dag ins not found")

func (t *dagInstanceFinder) GetRunningDagInstance(ctx context.Context, plan string, filters ...DagInsFilter) (*executor.ExecTaskStatusEx, error) {
	var zapFields []zap.Field
	for _, f := range filters {
		zapFields = append(zapFields, zap.String(f.ArgKey, strings.Join(f.ArgValues, ",")))
	}

	res, err := t.GetRunningDagInstances(ctx, []string{plan}, filters...)
	if err != nil {
		zapFields = append(zapFields, zap.Error(err))
		tilog.Error(ctx, fmt.Sprintf("query existed %s dag ins error", plan), zapFields...)
		return nil, fmt.Errorf("query existed %s dag ins error: %w", plan, err)
	}
	// abnormal condition
	if len(res) > 1 {
		zapFields = append(zapFields, zap.Any("dag ins", utils.JSON(res)))
		tilog.Error(ctx, fmt.Sprintf("two dag ins for a %s dag, there much be bugs", plan), zapFields...)
		return nil, fmt.Errorf("two dag ins for a %s dag, there much be bugs", plan)
	}
	// normal existed condition
	if len(res) == 1 {
		zapFields = append(zapFields, zap.Any("dag ins", utils.JSON(res)))
		tilog.Info(ctx, fmt.Sprintf("found the running %s dag", plan), zapFields...)
		for _, it := range res {
			return it, nil
		}
	}
	return nil, DagInsNotFoundErr
}

func (t *dagInstanceFinder) GetRunningDagInstances(ctx context.Context, plans []string, filters ...DagInsFilter) (executor.ExecTaskIdStatus, error) {
	return t.GetRunningDagInstancesWithFilterFunc(ctx, plans, matchAllFiltersFun, filters...)
}

func validateFilters(filterList []DagInsFilter) error {
	keyMap := map[string]bool{}
	for _, filter := range filterList {
		if _, ok := keyMap[filter.ArgKey]; ok {
			return fmt.Errorf("duplicate filter.ArgKey: %s", filter.ArgKey)
		}
		keyMap[filter.ArgKey] = true

		if filter.ArgKey == "" {
			return fmt.Errorf("invalid filter.ArgKey, empty is not allowed")
		}
		if len(filter.ArgValues) == 0 {
			return fmt.Errorf("invalid filter.ArgValues, no value found")
		}
		for _, val := range filter.ArgValues {
			if val == "" {
				return fmt.Errorf("invalid filter.ArgValues, empty is not allowed")
			}
		}
	}

	return nil
}

func matchAllFiltersFun(filters []DagInsFilter) executor.MatchFilter {
	mf := executor.NewMatchFilter()
	for _, filter := range filters {
		mf.Add(filter.ArgKey, filter.ArgValues)
	}
	return mf
}