package cdcaction

import (
	"context"
	"fmt"
	"strconv"

	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	opent "github.com/tidbcloud/devops/entity/operation"
	"github.com/tidbcloud/devops/pkg/service/operation/actions"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-agent/v1"
)

const (
	CDCAPIVERSIONV1  = "v1"
	CDCAPIVERSIONV2  = "v2"
	TICDC_COMPONENT  = "ticdc"
	NAMESPACE_PREFIX = "tidb"
)

var (
	ActionTiCDCPauseReconcile opent.ActionMetadata = opent.ActionMetadata{
		Name: "TicdcPauseReconcile",
		Type: opent.ActionTypeAuto}
)

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

func NewTiCDCPauseReconcile(comps *actions.Components) actions.AutoAction {
	return &ticdcPauseReconcile{
		comps: comps,
	}
}

func (p *ticdcPauseReconcile) Metadata() opent.ActionMetadata {
	return ActionTiCDCPauseReconcile
}

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

func (p *ticdcPauseReconcile) Start(ctx context.Context, instance *opent.Instance) (*actions.ActionResult, error) {
	ticdcClusterIDStr, err := p.getParam(instance)
	if err != nil {
		return actions.FailedActionResult(err), nil
	}
	ticdcClusterID, err := strconv.ParseUint(ticdcClusterIDStr, 10, 64)
	if err != nil {
		return actions.FailedActionResult(fmt.Errorf("invalid ticdc cluster id: %v", err)), nil
	}
	ticdcInfo, err := p.comps.DataflowCli.Ops.GetInternalTiCDCCluster(ctx, &dataflow.GetInternalTiCDCClusterRequest{
		TicdcClusterId: ticdcClusterID,
	})
	if err != nil {
		return actions.FailedActionResult(err), nil
	}
	switch ticdcInfo.OperatorVersion {
	case CDCAPIVERSIONV1:
		req := &apiv1.CdcPauseTCReq{
			TidbClusterId: strconv.FormatUint(ticdcInfo.TidbClusterId, 10),
			TcName:        ticdcInfo.Name,
		}
		_, err = p.comps.MgmtHTTP.CdcAgentV1(strconv.FormatUint(ticdcInfo.TidbClusterId, 10)).PauseReconcileV1(ctx, req)
		if err != nil {
			return actions.FailedActionResult(err), nil
		}
	case CDCAPIVERSIONV2:
		req := &apiv1.PauseClusterReq{
			Namespace:   fmt.Sprintf("tidb%s", strconv.FormatUint(ticdcInfo.TidbClusterId, 10)),
			ClusterName: "db", // FIX ME: here we use cluster id as cluster name, need to change to real cluster name
		}
		_, err = p.comps.MgmtHTTP.CdcAgentV2(strconv.FormatUint(ticdcInfo.TidbClusterId, 10)).PauseReconcileV2(ctx, req)
		if err != nil {
			return actions.FailedActionResult(err), nil
		}
	default:
		return nil, fmt.Errorf("unsupported cdc api version: %s", ticdcInfo.OperatorVersion)
	}
	return actions.SucceededActionResult(nil), nil // return succeeded directly
}

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

func (p *ticdcPauseReconcile) getParam(instance *opent.Instance) (string, error) {
	cdcClusterID, ok := instance.GetParam(opent.ParamCDCClusterID)
	if !ok {
		return "", fmt.Errorf("cdc cluster id param is not found")
	}
	return cdcClusterID, nil
}