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"
)

var (
	ActionTiCDCDeletePodsAndResumeReconcile opent.ActionMetadata = opent.ActionMetadata{
		Name: "TicdcDeletePodsAndResumeReconcile",
		Type: opent.ActionTypeAuto,
	}
)

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

func NewTiCDCDeletePodsAndResumeReconcile(comps *actions.Components) actions.AutoAction {
	return &ticdcDeletePodsAndResumeReconcile{
		comps: comps,
	}
}

func (p *ticdcDeletePodsAndResumeReconcile) Metadata() opent.ActionMetadata {
	return ActionTiCDCDeletePodsAndResumeReconcile
}

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

func (p *ticdcDeletePodsAndResumeReconcile) 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.CdcDeleteAstsAndResumeTCReq{
			TidbClusterId:  strconv.FormatUint(ticdcInfo.TidbClusterId, 10),
			TcName:         ticdcInfo.Name,
			OnlyDeleteAsts: false,
		}
		_, err = p.comps.MgmtHTTP.CdcAgentV1(strconv.FormatUint(ticdcInfo.TidbClusterId, 10)).DeletePodsAndResumeReconcileV1(ctx, req)
		if err != nil {
			return actions.FailedActionResult(err), nil
		}
	case CDCAPIVERSIONV2:
		req := &apiv1.DeleteGroupInstancesAndResumeClusterReq{
			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
			ComponentName: TICDC_COMPONENT,
			GroupName:     fmt.Sprintf("ticdc-%s", strconv.FormatUint(ticdcInfo.TidbClusterId, 10)),
		}
		_, err = p.comps.MgmtHTTP.CdcAgentV2(strconv.FormatUint(ticdcInfo.TidbClusterId, 10)).DeletePodsAndResumeReconcileV2(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 *ticdcDeletePodsAndResumeReconcile) Result(ctx context.Context, instance *opent.Instance) (*actions.ActionResult, error) {
	return actions.SucceededActionResult(nil), nil
}

func (p *ticdcDeletePodsAndResumeReconcile) 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
}