package del

import (
	"context"
	"fmt"

	"github.com/tidbcloud/dataflow-service/app/models/domain"
	ticdc_impl "github.com/tidbcloud/dataflow-service/app/models/ticdc/impl"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	"github.com/tidbcloud/dataflow-service/app/workflow/action/biz_changefeed"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

type ActionDeleteChangefeed struct {
	bizChangefeedRepo      repo.BizChangefeedRepo
	changefeedModelFactory *ticdc_impl.ChangefeedModelFactory
}

func NewDeleteChangefeedAction(
	bizChangefeedRepo repo.BizChangefeedRepo, changefeedModelFactory *ticdc_impl.ChangefeedModelFactory,
) *action.Wrapper {
	deleteAction := &ActionDeleteChangefeed{
		bizChangefeedRepo:      bizChangefeedRepo,
		changefeedModelFactory: changefeedModelFactory,
	}
	return action.MustNewActionWrapper(
		workflow.ActionBizChangefeedDeleteChangefeed,
		deleteAction,
		action.SetRunRetryFun(action.InvokeRunActionWithInfinitelyRetry), // Use infinite retry for transient network failures
	)
}

func (a *ActionDeleteChangefeed) Params() interface{} {
	return &biz_changefeed.CommonParams{}
}

func (a *ActionDeleteChangefeed) Run(ctx context.Context, logger *zap.Logger, params interface{}) error {
	commonParams := params.(*biz_changefeed.CommonParams)
	bizChangefeedEnt, bizErr := a.bizChangefeedRepo.Load(ctx, commonParams.BizChangefeedID)
	if bizErr != nil {
		return bizErr.Chain("load biz_changefeed entity")
	}
	kernelChangefeed, bizErr := a.queryKernelChangefeed(ctx, bizChangefeedEnt)
	if bizErr != nil {
		if bizErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			logger.Info("kernel changefeed is already deleted",
				zap.Uint64("biz_changefeed_id", commonParams.BizChangefeedID))
			return nil
		}
		return fmt.Errorf("query kernel changefeed error: %w", bizErr)
	}
	_, bizErr = a.changefeedModelFactory.GetModel().
		DeleteChangefeed(ctx, &domain.DeleteChangefeedReq{
			ChangefeedID: kernelChangefeed.ID.Uint64(),
		})
	return bizErr
}

func (a *ActionDeleteChangefeed) queryKernelChangefeed(
	ctx context.Context, bizChangefeedEnt *entity.BizChangefeed,
) (*entity.TiCDCChangefeed, bizerr.BizError) {
	bizType := domain.TiCDCClusterBizTypeChangefeed
	resp, err := a.changefeedModelFactory.GetModel().
		QueryChangefeed(ctx, &domain.QueryChangefeedReq{
			Filter: &domain.ListChangefeedFilter{
				BizType:     &bizType,
				BusinessIDs: []uint64{bizChangefeedEnt.ID.Uint64()},
			},
		})
	if err != nil {
		return nil, err
	}
	return &resp.ChangefeedEntity, nil
}