package create

import (
	"context"

	biz_changefeed_model "github.com/tidbcloud/dataflow-service/app/models/biz_changefeed"
	"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"
	"go.uber.org/zap"
)

type ActionFinalStep struct {
	bizChangefeedRepo   repo.BizChangefeedRepo
	bizChangefeedHooker biz_changefeed_model.BizChangefeedHooker
}

func NewFinalAction(
	bizChangefeedRepo repo.BizChangefeedRepo,
	bizChangefeedHooker biz_changefeed_model.BizChangefeedHooker,
) *action.Wrapper {
	finalAction := &ActionFinalStep{
		bizChangefeedRepo:   bizChangefeedRepo,
		bizChangefeedHooker: bizChangefeedHooker,
	}
	return action.MustNewActionWrapper(workflow.ActionBizChangefeedCreateFinalStep, finalAction)
}

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

func (a *ActionFinalStep) Run(ctx context.Context, logger *zap.Logger, params interface{}) (err error) {
	commonParams := params.(*biz_changefeed.CommonParams)
	bizChangefeedEnt, bizErr := a.bizChangefeedRepo.Load(ctx, commonParams.BizChangefeedID)
	if bizErr != nil {
		return bizErr.Chain("load biz_changefeed entity")
	}
	defer func() {
		if err == nil {
			a.bizChangefeedHooker.HandleBizChangefeedEvent(ctx, entity.CreateChangefeed, 0, bizChangefeedEnt)
		}
	}()

	if bizChangefeedEnt.OperationStatus == entity.BizChangefeedOperationStatusFailed {
		logger.Warn("biz_changefeed status is already failed, skip",
			zap.Uint64("biz_changefeed_id", commonParams.BizChangefeedID))
		return nil
	}

	bizChangefeedEnt.OperationStatus = entity.BizChangefeedOperationStatusNormal
	if bizErr = a.bizChangefeedRepo.Update(ctx, bizChangefeedEnt); bizErr != nil {
		return bizErr.Chain("update biz_changefeed status")
	}
	return nil
}