package cdcaction

import (
	"context"
	"fmt"
	"strconv"

	ccproto "github.com/tidbcloud/config-center/proto/config_center"
	dataflowng "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	opent "github.com/tidbcloud/devops/entity/operation"
	"github.com/tidbcloud/devops/pkg/service/operation/actions"
)

var (
	ActionRestartCDC opent.ActionMetadata = opent.ActionMetadata{
		Name: "RestartCDC",
		Type: opent.ActionTypeAuto,
	}
)

type restartCDC struct {
	actions.BaseAction
	comps *actions.Components
}

func NewRestartCDC(comps *actions.Components) actions.AutoAction {
	return &restartCDC{
		comps: comps,
	}
}

func (p *restartCDC) Metadata() opent.ActionMetadata {
	return ActionRestartCDC
}

func (p *restartCDC) PreCheck(ctx context.Context, instance *opent.Instance) error {
	_, _, err := p.getParam(instance)
	if err != nil {
		return err
	}

	return nil
}

func (p *restartCDC) Start(ctx context.Context, instance *opent.Instance) (*actions.ActionResult, error) {
	cdcClusterIDStr, clusterType, err := p.getParam(instance)
	if err != nil {
		return actions.FailedActionResult(err), nil
	}
	cdcClusterID, err := strconv.ParseUint(cdcClusterIDStr, 10, 64)
	if err != nil {
		return actions.FailedActionResult(err), nil
	}
	switch clusterType {
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_TICDC)]:
		_, err = p.comps.DataflowCli.Ops.RestartTiCDCCluster(ctx, &dataflow.RestartTiCDCClusterRequest{TicdcClusterId: cdcClusterID})
		if err != nil {
			return actions.FailedActionResult(err), nil
		}
		return actions.SucceededActionResult(nil), nil // return succeeded directly
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_TICDC_NG)]:
		_, err = p.comps.DataflowCli.NgOps.RestartTiCDCCluster(ctx, &dataflowng.RestartTiCDCClusterRequest{TicdcClusterId: cdcClusterID})
		if err != nil {
			return actions.FailedActionResult(err), nil
		}
		return actions.SucceededActionResult(nil), nil // return succeeded directly
	default:
		return actions.FailedActionResult(fmt.Errorf("invalid cluster_type param: %s", clusterType)), nil
	}

}

func (p *restartCDC) Result(ctx context.Context, instance *opent.Instance) (*actions.ActionResult, error) {
	return actions.SucceededActionResult(nil), nil
}

func (p *restartCDC) getParam(instance *opent.Instance) (string, string, error) {
	cdcClusterID, ok := instance.GetParam(opent.ParamCDCClusterID)
	if !ok {
		return "", "", fmt.Errorf("cdc cluster id param is not found")
	}
	clusterType, ok := instance.GetParam(opent.ParamCDCOperationClusterType)
	if !ok {
		return "", "", fmt.Errorf("cluster_type param is not found")
	}
	if clusterType != ccproto.ClusterType_name[int32(ccproto.ClusterType_TICDC)] && clusterType != ccproto.ClusterType_name[int32(ccproto.ClusterType_TICDC_NG)] {
		return "", "", fmt.Errorf("invalid cluster_type param: %s", clusterType)
	}
	return cdcClusterID, clusterType, nil
}