package create

import (
	"context"
	"fmt"

	appconfig "github.com/tidbcloud/dataflow-service/app/config"
	biz_changefeed_impl "github.com/tidbcloud/dataflow-service/app/models/biz_changefeed/impl"
	"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"
	internal_metrics "github.com/tidbcloud/dataflow-service/internal/metrics"
	errhandler "github.com/tidbcloud/dataflow-service/pkg/error-handler"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

// After retry times reach max, we will set changefeed to failed and end the dag.
// Work with InvokeRunActionWithInfinitelyRetry, the max timeout will be 20 * 15s = 300s = 5m
// Increase max timeout to 5m since in some cases cdc pod take minutes to ready to be called after scaling/creation done
const createChangefeedMaxRetryNum = 20

type ChangefeedCreateParams struct {
	BizChangefeedID uint64 `json:"biz_changefeed_id"`
	createRetryNum  int
}

type ActionCreateChangefeed struct {
	downstreamFactory      *biz_changefeed_impl.DownstreamAdaptorFactory
	bizChangefeedRepo      repo.BizChangefeedRepo
	changefeedModelFactory *ticdc_impl.ChangefeedModelFactory
	coreCli                grpcli.CoreGrpcClient
}

func NewCreateChangefeedAction(
	downstreamFactory *biz_changefeed_impl.DownstreamAdaptorFactory,
	bizChangefeedRepo repo.BizChangefeedRepo,
	changefeedModelFactory *ticdc_impl.ChangefeedModelFactory,
	coreCli grpcli.CoreGrpcClient,
) *action.Wrapper {
	createAction := &ActionCreateChangefeed{
		downstreamFactory:      downstreamFactory,
		bizChangefeedRepo:      bizChangefeedRepo,
		changefeedModelFactory: changefeedModelFactory,
		coreCli:                coreCli,
	}
	return action.MustNewActionWrapper(
		workflow.ActionBizChangefeedCreateChangefeed,
		createAction,
		action.SetRunRetryFun(action.InvokeRunActionWithInfinitelyRetry))
}

func (a *ActionCreateChangefeed) Params() interface{} {
	return &ChangefeedCreateParams{}
}

func (a *ActionCreateChangefeed) Run(ctx context.Context, logger *zap.Logger, params interface{}) error {
	createParams := params.(*ChangefeedCreateParams)
	if createParams.createRetryNum >= createChangefeedMaxRetryNum {
		logger.Warn("creating changefeed already exceeded max retry number",
			zap.Uint64("biz_changefeed_id", createParams.BizChangefeedID))
		return nil
	}

	bizChangefeedEnt, bizErr := a.bizChangefeedRepo.Load(ctx, createParams.BizChangefeedID)
	if bizErr != nil {
		return bizErr.Chain("load biz_changefeed entity %d", createParams.BizChangefeedID)
	}
	downstreamAdaptor, bizErr := a.downstreamFactory.Get(
		domain.TiCDCChangefeedDownStreamType2ConnectorType(bizChangefeedEnt.DownstreamType))
	if bizErr != nil {
		return bizErr.Chain("get downstream adaptor")
	}
	cluster, bizErr := a.coreCli.GetAvailableClusterByID(ctx, bizChangefeedEnt.ClusterID.Uint64())
	if bizErr != nil {
		return bizErr.Chain("get cluster")
	}
	changefeedConfig, bizErr := downstreamAdaptor.AssembleChangefeedConfig(
		ctx, cluster, bizChangefeedEnt)
	if bizErr != nil {
		return bizErr.Chain("assemble changefeed config")
	}

	return a.createChangefeed(ctx, logger, bizChangefeedEnt, changefeedConfig, createParams)
}

func (a *ActionCreateChangefeed) createChangefeed(
	ctx context.Context,
	logger *zap.Logger,
	bizChangefeedEnt *entity.BizChangefeed,
	changefeedConfig *domain.TiCDCChangefeedConfig,
	createParams *ChangefeedCreateParams,
) error {
	req := &domain.CreateChangefeedReq{
		TiCDCClusterID:     bizChangefeedEnt.TiCDCClusterID.Uint64(),
		KernelChangefeedID: bizChangefeedEnt.GenKernelChangefeedID(),
		DownStreamType:     bizChangefeedEnt.DownstreamType,
		Config:             changefeedConfig,
		BizType:            domain.TiCDCClusterBizTypeChangefeed,
		BusinessID:         bizChangefeedEnt.ID.Uint64(),
	}
	resp, bizErr := a.changefeedModelFactory.GetModel().CreateChangefeed(ctx, req)
	if bizErr != nil {
		if bizErr.RawErrCode() == appconfig.ErrCodeCreateKernelChangefeedFailed {
			logger.Error("create kernel changefeed error",
				zap.Uint64("biz_changefeed_id", bizChangefeedEnt.ID.Uint64()),
				zap.Error(bizErr))
			return a.handleKernelError(ctx, bizChangefeedEnt, createParams, bizErr)
		} else {
			return bizErr.Chain("create changefeed")
		}
	}

	logger.Info("create changefeed successfully",
		zap.Uint64("changefeed_id", resp.ChangefeedID),
		zap.Uint64("ticdc_cluster_id", resp.TiCDCClusterID),
		zap.Uint64("biz_changefeed_id", bizChangefeedEnt.ID.Uint64()))
	return nil
}

func (a *ActionCreateChangefeed) handleKernelError(
	ctx context.Context,
	bizChangefeedEnt *entity.BizChangefeed,
	createParams *ChangefeedCreateParams,
	err bizerr.BizError,
) error {
	currentRetryNum := createParams.createRetryNum + 1
	if currentRetryNum < createChangefeedMaxRetryNum {
		createParams.createRetryNum = currentRetryNum
		return err
	}

	bizChangefeedEnt.OperationStatus = entity.BizChangefeedOperationStatusFailed
	bizChangefeedEnt.ErrMsg = err.ErrMsg()
	if bizErr := a.bizChangefeedRepo.Update(ctx, bizChangefeedEnt); bizErr != nil {
		return bizErr.Chain("record kernel changefeed error msg: %s", err.Error())
	}

	internal_metrics.ChangefeedCreateResultCounter.WithLabelValues(
		fmt.Sprintf("%d", bizChangefeedEnt.ClusterID.Uint64()),
		fmt.Sprintf("%d", bizChangefeedEnt.ClusterID.Uint64()),
		fmt.Sprintf("changefeed-%d", bizChangefeedEnt.ID.Uint64()),
		internal_metrics.Dedicate,
		internal_metrics.ChangefeedCreateResultFailed,
		err.ErrMsg(),
		errhandler.ChangefeedErrType(err.ErrMsg())).Inc()

	createParams.createRetryNum = currentRetryNum
	return nil
}