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"
	ccent "github.com/tidbcloud/devops/entity/cloudconfig"
	opent "github.com/tidbcloud/devops/entity/operation"
	"github.com/tidbcloud/devops/pkg/service/operation/actions"
)

var (
	ActionRestartChangefeed opent.ActionMetadata = opent.ActionMetadata{
		Name: "RestartChangefeed",
		Type: opent.ActionTypeAuto,
	}
)

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

func NewRestartChangefeed(comps *actions.Components) actions.AutoAction {
	return &restartChangefeed{
		comps: comps,
	}
}

func (p *restartChangefeed) Metadata() opent.ActionMetadata {
	return ActionRestartChangefeed
}

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

	return nil
}

func (p *restartChangefeed) Start(ctx context.Context, instance *opent.Instance) (*actions.ActionResult, error) {
	changefeedIDStr, clusterType, err := p.getParam(instance)
	if err != nil {
		return actions.FailedActionResult(err), nil
	}
	switch clusterType {
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_CHANGEFEED)]:
		changefeedID, err := strconv.ParseUint(changefeedIDStr, 10, 64)
		if err != nil {
			return actions.FailedActionResult(err), nil
		}
		_, err = p.comps.DataflowCli.Ops.RestartChangefeed(ctx, &dataflow.RestartChangefeedRequest{ChangefeedId: changefeedID})
		if err != nil {
			return actions.FailedActionResult(err), nil
		}
		return actions.SucceededActionResult(nil), nil // return succeeded directly
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_CHANGEFEED_NG)]:
		changefeedID, err := strconv.ParseUint(changefeedIDStr, 10, 64)
		if err != nil {
			return actions.FailedActionResult(err), nil
		}
		_, err = p.comps.DataflowCli.NgOps.RestartChangefeed(ctx, &dataflowng.RestartChangefeedRequest{ChangefeedId: changefeedID})
		if err != nil {
			return actions.FailedActionResult(err), nil
		}
		return actions.SucceededActionResult(nil), nil // return succeeded directly
	case string(ccent.ClusterTypeChangefeedEssential):
		_, err = p.comps.ServerlessCli.RestartChangefeed(ctx, changefeedIDStr)
		if err != nil {
			return actions.FailedActionResult(err), nil
		}
		return actions.SucceededActionResult(nil), nil
	default:
		return actions.FailedActionResult(fmt.Errorf("invalid cluster_type param: %s", clusterType)), nil
	}
}

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

func (p *restartChangefeed) getParam(instance *opent.Instance) (string, string, error) {
	changefeedID, ok := instance.GetParam(opent.ParamChangefeedID)
	if !ok {
		return "", "", fmt.Errorf("changefeed_id param is not found")
	}
	clusterType, ok := instance.GetParam(opent.ParamCDCOperationClusterType)
	if !ok {
		return "", "", fmt.Errorf("cluster_type param is not found")
	}
	if !ccent.IsSupportedChangefeedClusterType(clusterType) {
		return "", "", fmt.Errorf("invalid cluster_type param: %s", clusterType)
	}
	return changefeedID, clusterType, nil
}