package create

import (
	"context"
	"time"

	"github.com/avast/retry-go/v4"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	"github.com/tidbcloud/dataflow-service/app/internal/feature_gate"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	restoreRepo "github.com/tidbcloud/dataflow-service/app/repo/restore"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	"github.com/tidbcloud/dataflow-service/app/workflow/action/restore/base"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
)

type checkAndCreateClusterParam struct {
	RestoreID uint64 `json:"restore_id"`
}

type CheckAndCreateCluster struct {
	base.BaseAction
}

func NewCheckAndCreateCluster(
	transProvider db.TransProvider,
	restoreRepo restoreRepo.RestoreRepo,
	coreCli grpcli.CoreGrpcClient,
	mgmtSvcCli grpcli.MgmtSvcGrpcClient,
	featureGate feature_gate.FeatureGate,
) *CheckAndCreateCluster {
	return &CheckAndCreateCluster{
		BaseAction: base.NewBaseAction(transProvider, coreCli, mgmtSvcCli, restoreRepo, featureGate),
	}
}

func (a *CheckAndCreateCluster) Name() string {
	return workflow.ActionCheckAndCreateClusterForRestoring
}

func (a *CheckAndCreateCluster) Run(ctx run.ExecuteContext, params interface{}) error {
	logger := action.GetLogger(ctx).Named(a.Name())
	input := params.(*checkAndCreateClusterParam)
	logger.Info("action run start")
	err := action.InvokeRunActionWithRetry(ctx.Context(), logger, func() error {
		return a.createCluster(ctx.Context(), input)
	})
	if err != nil {
		logger.Error("action run error", zap.Error(err))
		return err
	}
	logger.Info("action run finished")
	return nil
}

func (a *CheckAndCreateCluster) createCluster(
	ctx context.Context, input *checkAndCreateClusterParam,
) bizerr.BizError {
	logger := log.GetBaseLogger(ctx)
	restoreEntity, err := a.RestoreRepo.Load(ctx, input.RestoreID)
	if err != nil {
		logger.Warn("load restore entity error", zap.Error(err))
		return err
	}
	if restoreEntity.ClusterID != 0 {
		logger.Info("cluster already created", zap.Any("cluster_id", restoreEntity.ClusterID))
		return nil
	}
	//nolint:nestif
	if restoreEntity.RequestArgsV2 != "" {
		createClusterReqV2 := &mgmtv1.RestoreDedicatedClusterRequest{}
		if err := protojson.Unmarshal([]byte(restoreEntity.RequestArgsV2), createClusterReqV2); err != nil {
			return bizerr.WrapError(err, "failed to restore request args")
		}
		createClusterReqV2.RestoreId = uint64(restoreEntity.ID)
		createClusterResp, err := a.MgmtSvcCli.RestoreDedicatedCluster(ctx, createClusterReqV2)
		if err != nil {
			if err.RawErrCode() == bizerr.ErrCodeRecordExist {
				latestRestoreEntity, loadErr := a.RestoreRepo.Load(ctx, input.RestoreID)
				if loadErr != nil {
					logger.Warn("load restore entity error", zap.Error(loadErr))
					return loadErr
				}
				if latestRestoreEntity.ClusterID != 0 {
					logger.Info("ignore duplicate create cluster request",
						zap.Uint64("restore_id", input.RestoreID),
						zap.Uint64("cluster_id", latestRestoreEntity.ClusterID.Uint64()),
					)
					return nil
				}
				logger.Warn("duplicate create cluster request but cluster id is still empty",
					zap.Uint64("restore_id", input.RestoreID), zap.Error(err))
			}
			logger.Warn("create cluster error", zap.Error(err))
			return err
		}
		if createClusterResp.ClusterId == 0 {
			logger.Warn("create cluster resp id is empty")
			return bizerr_wrapper.NewInternalErr("create cluster resp id is empty")
		}
		restoreEntity.ClusterID = entity.ID(createClusterResp.ClusterId)
		if err := a.RestoreRepo.Save(ctx, restoreEntity); err != nil {
			logger.Warn("save restore entity error", zap.Error(err))
			return err
		}
	} else {
		logger.Warn("both restore request and request v2 args is empty")
		return bizerr_wrapper.NewInternalErr("both restore request and request v2 args is empty")
	}
	return nil
}

//nolint:gomnd
func (a *CheckAndCreateCluster) RunAfter(ctx run.ExecuteContext, params interface{}) error {
	logger := action.GetLogger(ctx).Named(a.Name())
	input := params.(*checkAndCreateClusterParam)
	logger.Info("action run after start")
	err := retry.Do(func() error {
		return a.checkClusterIsReady(ctx.Context(), input)
	},
		retry.Context(ctx.Context()),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(time.Second*15),
		retry.Attempts(0),
		retry.OnRetry(func(n uint, err error) {
			logger.Info("action runAfter error", zap.Uint("retry_times", n))
		}),
		retry.LastErrorOnly(true),
		retry.RetryIf(func(err error) bool {
			return err != action.StopRetry
		}),
	)
	if err != nil {
		logger.Error("action run after error", zap.Error(err))
		return err
	}
	logger.Info("action run after finished")
	return nil
}

func (a *CheckAndCreateCluster) checkClusterIsReady(
	ctx context.Context, input *checkAndCreateClusterParam,
) bizerr.BizError {
	logger := log.GetBaseLogger(ctx)
	restoreEntity, err := a.RestoreRepo.Load(ctx, input.RestoreID)
	if err != nil {
		logger.Warn("load restore entity error", zap.Error(err))
		return err
	}
	if restoreEntity.ClusterID == 0 {
		logger.Error("cluster is not created", zap.Any("cluster_id", restoreEntity.ClusterID))
		return action.StopRetry
	}
	cluster, err := a.MgmtSvcCli.GetAnyDedicatedClusterBasicByID(ctx, uint64(restoreEntity.ClusterID))
	if err != nil {
		logger.Warn("get cluster error", zap.Error(err))
		return err
	}
	dc := cluster.GetDedicatedCluster()
	if dc.GetState() != dedicatedv1.Cluster_AVAILABLE {
		logger.Info("cluster state is not available",
			zap.Any("cluster_id", restoreEntity.ClusterID),
			zap.Any("cluster_state", dc.GetState()),
		)
		return bizerr_wrapper.NewInternalErr("cluster state is not available")
	}
	restoreEntity.RequestArgsV2 = ""
	if err := a.RestoreRepo.Save(ctx, restoreEntity); err != nil {
		logger.Warn("save restore entity error", zap.Error(err))
		return err
	}
	return nil
}

func (a *CheckAndCreateCluster) ParameterNew() interface{} {
	return &checkAndCreateClusterParam{}
}