package premium_dm

import (
	"context"

	"github.com/hashicorp/go-multierror"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service-ng/pkg/bizerr-wrapper"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

type premiumDMPrecheckDAO interface {
	CreatePrecheck(ctx context.Context, precheck *entity.PremiumDMPrecheck) bizerr.BizError
	ListPrechecks(
		ctx context.Context, filter *entity.PremiumDMPrecheckFilter,
	) ([]*entity.PremiumDMPrecheck, bizerr.BizError)
	GetPrecheck(ctx context.Context, clusterID, id string) (*entity.PremiumDMPrecheck, bizerr.BizError)
	UpdatePrecheck(ctx context.Context, id string, updates map[string]any) bizerr.BizError
	CreatePrecheckItem(ctx context.Context, item *entity.PremiumDMPrecheckItem) bizerr.BizError
	UpdatePrecheckItem(ctx context.Context, id string, updates map[string]any) bizerr.BizError
}

type PremiumDMPrecheckWorker struct {
	dao         premiumDMPrecheckDAO
	taskWorker  *PremiumDMTaskWorker
	runPrecheck func(ctx context.Context, precheck *entity.PremiumDMPrecheck) error
}

func NewPremiumDMPrecheckWorker(dao *PremiumDMTaskDAO, taskWorker *PremiumDMTaskWorker) *PremiumDMPrecheckWorker {
	worker := &PremiumDMPrecheckWorker{dao: dao, taskWorker: taskWorker}
	worker.runPrecheck = worker.runSinglePrecheck
	return worker
}

func (w *PremiumDMPrecheckWorker) CreatePrecheck(
	ctx context.Context, precheck *entity.PremiumDMPrecheck,
) bizerr.BizError {
	if precheck == nil {
		return bizerr_wrapper.NewInternalErr("precheck is nil")
	}
	if precheck.ID == "" {
		precheck.ID = entity.GenerateDMIDWithPrefix("pck")
	}
	if precheck.Stage == "" {
		precheck.Stage = string(entity.PremiumDMPrecheckStagePending)
	}
	if precheck.TiDBID == "" {
		return bizerr_wrapper.NewInternalErr("tidb id is empty")
	}
	return w.dao.CreatePrecheck(ctx, precheck)
}

func (w *PremiumDMPrecheckWorker) Run(ctx context.Context) error {
	logger := GetWorkerLoggerFromContext(ctx)
	prechecks, bizErr := w.listPendingPrechecks(ctx)
	if bizErr != nil {
		logger.Error("failed to list pending prechecks", zap.Error(bizErr))
		return bizErr
	}
	if len(prechecks) == 0 {
		return nil
	}

	runPrecheck := w.runPrecheck
	if runPrecheck == nil {
		runPrecheck = w.runSinglePrecheck
	}

	var errs error
	for _, precheck := range prechecks {
		if err := runPrecheck(ctx, precheck); err != nil {
			logger.Error("failed to run precheck",
				zap.String("precheck_id", precheck.ID),
				zap.String("cluster_id", precheck.TiDBID),
				zap.Error(err),
			)
			errs = multierror.Append(errs, err)
		}
	}
	return errs
}

func (w *PremiumDMPrecheckWorker) listPendingPrechecks(
	ctx context.Context,
) ([]*entity.PremiumDMPrecheck, bizerr.BizError) {
	return w.dao.ListPrechecks(ctx, &entity.PremiumDMPrecheckFilter{
		Stages:  []string{string(entity.PremiumDMPrecheckStagePending)},
		OrderBy: "created_at ASC",
	})
}