package cloudconfig

import (
	"fmt"
	"strconv"

	"github.com/gin-gonic/gin"
	userapiv1 "github.com/tidbcloud/aisupport/packages/sdk-go/contracts/api/core-server/user/v1"
	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"
	"github.com/tidbcloud/devops/api"
	ccapi "github.com/tidbcloud/devops/api/cloudconfig"
	ccent "github.com/tidbcloud/devops/entity/cloudconfig"
	opent "github.com/tidbcloud/devops/entity/operation"
	ctrl "github.com/tidbcloud/devops/pkg/controller"
	"github.com/tidbcloud/devops/pkg/infra/consts"
	ccsvcapi "github.com/tidbcloud/devops/pkg/service/cloudconfig/api"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/cvt"
	opsvc "github.com/tidbcloud/devops/pkg/service/operation"
	"github.com/tidbcloud/devops/pkg/service/operation/actions/cdcaction"
	"github.com/tidbcloud/devops/pkg/service/operation/actions/opsaction"
	opsvcins "github.com/tidbcloud/devops/pkg/service/operation/instance"
)

const (
	CDC3LAYERARCH   = "3layer"
	CDCAPIVERSIONV1 = "v1"
	CDCAPIVERSIONV2 = "v2"
)

func validateChangefeedOperationClusterType(clusterType string) error {
	if !ccent.IsSupportedChangefeedClusterType(clusterType) {
		return fmt.Errorf("unsupported cluster_type: %s", clusterType)
	}
	return nil
}

// ListInternalChangefeeds
//
//	@Id				CloudConfigControllerListInternalChangefeeds
//	@Description	list business changefeeds of a cluster
//	@Tags			CloudConfigController
//	@Accept			json
//	@Produce		json
//	@Param			cluster_id		path		string	true	"cluster id"
//	@Param			cluster_type	path		string	true	"cluster type"
//	@Success		200				{object}	ccapi.ListInternalChangefeedsResp
//	@Failure		default			{object}	api.BaseResp
//	@Router			/api/v1/devops/ticdc-meta/{cluster_type}/internal-changefeeds/{cluster_id} [get]
func (c *CloudConfigController) ListInternalChangefeeds(ctx *gin.Context) {
	clusterIDStr := ctx.Param("cluster_id")
	if clusterIDStr == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_id is required"), api.ErrCodeInvalidArgument)
		return
	}
	clusterType := ctx.Param("cluster_type")
	if clusterType == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_type is required"), api.ErrCodeInvalidArgument)
		return
	}
	clusterID, err := strconv.ParseUint(clusterIDStr, 10, 64)
	if err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	if err := ctrl.CheckClusterScope(ctx, c.coreSvcCli, clusterID); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodePermissionDenied)
		return
	}

	switch clusterType {
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_DEDICATED_TIDB)],
		ccproto.ClusterType_name[int32(ccproto.ClusterType_LOGICAL_CLUSTER)],
		ccproto.ClusterType_name[int32(ccproto.ClusterType_SHARED_POOL)],
		string(ccent.ClusterTypeChangefeedEssential):
	default:
		ctrl.FailWithCode(ctx, fmt.Errorf("unsupported cluster_type: %s", clusterType), api.ErrCodeInvalidArgument)
		return
	}

	resp, err := c.svc.ListInternalChangefeeds(ctx, ccsvcapi.ListInternalChangefeedsParam{
		TiDBClusterID: clusterID,
		ClusterType:   clusterType,
	})
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}
	ctrl.Success(ctx, resp)
}

// GetInternalChangefeed
//
//	@Id				CloudConfigControllerGetInternalChangefeed
//	@Description	get one changefeed by kernel/internal changefeed id
//	@Tags			CloudConfigController
//	@Accept			json
//	@Produce		json
//	@Param			changefeed_id	path		string	true	"kernel/internal changefeed id"
//	@Param			cluster_type	path		string	true	"cluster type"
//	@Success		200				{object}	ccapi.GetInternalChangefeedResp
//	@Failure		default			{object}	api.BaseResp
//	@Router			/api/v1/devops/ticdc-meta/{cluster_type}/internal-changefeed/{changefeed_id} [get]
func (c *CloudConfigController) GetInternalChangefeed(ctx *gin.Context) {
	changefeedIDStr := ctx.Param("changefeed_id")
	if changefeedIDStr == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("changefeed id is required"), api.ErrCodeInvalidArgument)
		return
	}
	clusterType := ctx.Param("cluster_type")
	if clusterType == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_type is required"), api.ErrCodeInvalidArgument)
		return
	}
	param := ccsvcapi.GetInternalChangefeedParam{ClusterType: clusterType}
	switch clusterType {
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_CHANGEFEED)],
		ccproto.ClusterType_name[int32(ccproto.ClusterType_CHANGEFEED_NG)]:
		changefeedID, err := strconv.ParseUint(changefeedIDStr, 10, 64)
		if err != nil {
			ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
			return
		}
		param.ChangefeedID = changefeedID
	case string(ccent.ClusterTypeChangefeedEssential):
		param.ChangefeedIDString = changefeedIDStr
	default:
		ctrl.FailWithCode(ctx, fmt.Errorf("unsupported cluster_type: %s", clusterType), api.ErrCodeInvalidArgument)
		return
	}

	result, err := c.svc.GetInternalChangefeed(ctx, param)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}
	if result == nil || result.Changefeed == nil || result.TiDBClusterID == 0 {
		ctrl.Fail(ctx, fmt.Errorf("changefeed %s has incomplete unified metadata", changefeedIDStr))
		return
	}
	if err := ctrl.CheckClusterScope(ctx, c.coreSvcCli, result.TiDBClusterID); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodePermissionDenied)
		return
	}

	ctrl.Success(ctx, &ccapi.GetInternalChangefeedResp{Changefeed: result.Changefeed})
}

// GetInternalTiCDCCluster
//
//	@Id				CloudConfigControllerGetInternalTiCDCCluster
//	@Description	get one ticdc cluster by ticdc cluster id
//	@Tags			CloudConfigController
//	@Accept			json
//	@Produce		json
//	@Param			ticdc_cluster_id	path		string	true	"ticdc cluster id"
//	@Param			cluster_type		path		string	true	"ticdc cluster type"
//	@Success		200					{object}	ccapi.GetInternalTiCDCClusterResp
//	@Failure		default				{object}	api.BaseResp
//	@Router			/api/v1/devops/ticdc-meta/{cluster_type}/internal-ticdc/{ticdc_cluster_id} [get]
func (c *CloudConfigController) GetInternalTiCDCCluster(ctx *gin.Context) {
	ticdcClusterIDStr := ctx.Param("ticdc_cluster_id")
	if ticdcClusterIDStr == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("ticdc cluster id is required"), api.ErrCodeInvalidArgument)
		return
	}
	ticdcClusterID, err := strconv.ParseUint(ticdcClusterIDStr, 10, 64)
	if err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	clusterType := ctx.Param("cluster_type")
	if clusterType == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_type is required"), api.ErrCodeInvalidArgument)
		return
	}
	switch clusterType {
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_TICDC)]:
		ticdcCluster, err := c.svc.GetInternalTiCDCCluster(ctx, &dataflow.GetInternalTiCDCClusterRequest{TicdcClusterId: ticdcClusterID})
		if err != nil {
			ctrl.Fail(ctx, err)
			return
		}

		resp := &ccapi.GetInternalTiCDCClusterResp{TiCDCCluster: cvt.From.Dataflow.TiCDCCluster(ticdcCluster)}

		ctrl.Success(ctx, resp)
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_TICDC_NG)]:
		ticdcCluster, err := c.svc.GetInternalNgTiCDCCluster(ctx, &dataflowng.GetInternalTiCDCClusterRequest{TicdcClusterId: ticdcClusterID})
		if err != nil {
			ctrl.Fail(ctx, err)
			return
		}
		resp := &ccapi.GetInternalTiCDCClusterResp{TiCDCCluster: cvt.From.DataflowNg.TiCDCCluster(ticdcCluster)}
		ctrl.Success(ctx, resp)
	default:
		ctrl.FailWithCode(ctx, fmt.Errorf("unsupported cluster_type: %s", clusterType), api.ErrCodeInvalidArgument)
	}

}

// RestartTiCDCCluster
//
//	@Id				CloudConfigControllerRestartTiCDCCluster
//	@Description	restart one ticdc cluster by ticdc cluster id
//	@Tags			CloudConfigController
//	@Accept			json
//	@Produce		json
//	@Param			ticdc_cluster_id	path		string	true	"ticdc cluster id"
//	@Param			cluster_type		path		string	true	"cluster type"
//	@Success		200					{object}	ccapi.OperationResp
//	@Failure		default				{object}	api.BaseResp
//	@Router			/api/v1/devops/ticdc-meta/{cluster_type}/internal-ticdc/{ticdc_cluster_id} [post]
func (c *CloudConfigController) RestartTiCDCCluster(ctx *gin.Context) {
	ticdcClusterIDStr := ctx.Param("ticdc_cluster_id")
	if ticdcClusterIDStr == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("ticdc cluster id is required"), api.ErrCodeInvalidArgument)
		return
	}

	clusterType := ctx.Param("cluster_type")
	if clusterType == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_type is required"), api.ErrCodeInvalidArgument)
		return
	}
	if clusterType != ccproto.ClusterType_name[int32(ccproto.ClusterType_TICDC)] && clusterType != ccproto.ClusterType_name[int32(ccproto.ClusterType_TICDC_NG)] {
		ctrl.FailWithCode(ctx, fmt.Errorf("unsupported cluster_type: %s", clusterType), api.ErrCodeInvalidArgument)
		return
	}

	user, ok := ctx.Get(consts.CtxUserProfile)
	if !ok {
		ctrl.FailWithCode(ctx, fmt.Errorf("user profile not found"), api.ErrCodeUnauthenticated)
		return
	}
	userProfile := user.(*userapiv1.PrincipalProfile)
	createOPparam := opsvc.CreateInstanceParam{
		Type:      opent.OperationTargetUndefined,
		Operation: "RestartCDC",
		Applicant: userProfile.UserEmail,
		Reason:    "",
		Params: map[opent.ParamName]string{
			opent.ParamCDCClusterID:            ticdcClusterIDStr,
			opent.ParamCDCOperationClusterType: clusterType,
		},
		Actions: []opsvcins.CreateInstanceActionParam{
			{
				Name:        cdcaction.ActionRestartCDC.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
		},
	}

	insID, err := c.opSvc.CreateInstance(ctx, createOPparam)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	resp := &ccapi.OperationResp{InstanceID: insID}
	ctrl.Success(ctx, resp)
}

// RestartChangefeed
//
//	@Id				CloudConfigControllerRestartChangefeed
//	@Description	restart one changefeed by changefeed id
//	@Tags			CloudConfigController
//	@Accept			json
//	@Produce		json
//	@Param			changefeed_id	path		string	true	"changefeed id"
//	@Param			cluster_type	path		string	true	"cluster type"
//	@Success		200				{object}	ccapi.OperationResp
//	@Failure		default			{object}	api.BaseResp
//	@Router			/api/v1/devops/ticdc-meta/{cluster_type}/internal-changefeed/{changefeed_id} [post]
func (c *CloudConfigController) RestartChangefeed(ctx *gin.Context) {
	changefeedIDStr := ctx.Param("changefeed_id")
	if changefeedIDStr == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("changefeed id is required"), api.ErrCodeInvalidArgument)
		return
	}
	clusterType := ctx.Param("cluster_type")
	if clusterType == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_type is required"), api.ErrCodeInvalidArgument)
		return
	}
	if err := validateChangefeedOperationClusterType(clusterType); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	user, ok := ctx.Get(consts.CtxUserProfile)
	if !ok {
		ctrl.FailWithCode(ctx, fmt.Errorf("user profile not found"), api.ErrCodeUnauthenticated)
		return
	}
	userProfile := user.(*userapiv1.PrincipalProfile)

	createOPparam := opsvc.CreateInstanceParam{
		Type:      opent.OperationTargetUndefined,
		Operation: "RestartChangefeed",
		Applicant: userProfile.UserEmail,
		Reason:    "",
		Params: map[opent.ParamName]string{
			opent.ParamChangefeedID:            changefeedIDStr,
			opent.ParamCDCOperationClusterType: clusterType,
		},
		Actions: []opsvcins.CreateInstanceActionParam{
			{
				Name:        cdcaction.ActionRestartChangefeed.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
		},
	}

	insID, err := c.opSvc.CreateInstance(ctx, createOPparam)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	resp := &ccapi.OperationResp{InstanceID: insID}
	ctrl.Success(ctx, resp)
}

// PauseChangefeed
//
//	@Id				CloudConfigControllerPauseChangefeed
//	@Description	pause one changefeed by changefeed id
//	@Tags			CloudConfigController
//	@Accept			json
//	@Produce		json
//	@Param			changefeed_id	path		string	true	"changefeed id"
//	@Param			cluster_type	path		string	true	"cluster type"
//	@Success		200				{object}	ccapi.OperationResp
//	@Failure		default			{object}	api.BaseResp
//	@Router			/api/v1/devops/ticdc-meta/{cluster_type}/internal-changefeed/pause/{changefeed_id} [post]
func (c *CloudConfigController) PauseChangefeed(ctx *gin.Context) {
	changefeedIDStr := ctx.Param("changefeed_id")
	if changefeedIDStr == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("changefeed id is required"), api.ErrCodeInvalidArgument)
		return
	}

	clusterType := ctx.Param("cluster_type")
	if clusterType == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_type is required"), api.ErrCodeInvalidArgument)
		return
	}
	if err := validateChangefeedOperationClusterType(clusterType); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	user, ok := ctx.Get(consts.CtxUserProfile)
	if !ok {
		ctrl.FailWithCode(ctx, fmt.Errorf("user profile not found"), api.ErrCodeUnauthenticated)
		return
	}
	userProfile := user.(*userapiv1.PrincipalProfile)

	createOPparam := opsvc.CreateInstanceParam{
		Type:      opent.OperationTargetUndefined,
		Operation: "PauseChangefeed",
		Applicant: userProfile.UserEmail,
		Reason:    "",
		Params: map[opent.ParamName]string{
			opent.ParamChangefeedID:            changefeedIDStr,
			opent.ParamCDCOperationClusterType: clusterType,
		},
		Actions: []opsvcins.CreateInstanceActionParam{
			{
				Name:        cdcaction.ActionPauseChangefeed.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
		},
	}

	insID, err := c.opSvc.CreateInstance(ctx, createOPparam)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	resp := &ccapi.OperationResp{InstanceID: insID}
	ctrl.Success(ctx, resp)
}

// ResumeChangefeed
//
//	@Id				CloudConfigControllerResumeChangefeed
//	@Description	resume one changefeed by changefeed id
//	@Tags			CloudConfigController
//	@Accept			json
//	@Produce		json
//	@Param			changefeed_id	path		string	true	"changefeed id"
//	@Param			cluster_type	path		string	true	"cluster type"
//	@Success		200				{object}	ccapi.OperationResp
//	@Failure		default			{object}	api.BaseResp
//	@Router			/api/v1/devops/ticdc-meta/{cluster_type}/internal-changefeed/resume/{changefeed_id} [post]
func (c *CloudConfigController) ResumeChangefeed(ctx *gin.Context) {
	changefeedIDStr := ctx.Param("changefeed_id")
	if changefeedIDStr == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("changefeed id is required"), api.ErrCodeInvalidArgument)
		return
	}
	clusterType := ctx.Param("cluster_type")
	if clusterType == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("cluster_type is required"), api.ErrCodeInvalidArgument)
		return
	}
	if err := validateChangefeedOperationClusterType(clusterType); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	user, ok := ctx.Get(consts.CtxUserProfile)
	if !ok {
		ctrl.FailWithCode(ctx, fmt.Errorf("user profile not found"), api.ErrCodeUnauthenticated)
		return
	}
	userProfile := user.(*userapiv1.PrincipalProfile)

	createOPparam := opsvc.CreateInstanceParam{
		Type:      opent.OperationTargetUndefined,
		Operation: "ResumeChangefeed",
		Applicant: userProfile.UserEmail,
		Reason:    "",
		Params: map[opent.ParamName]string{
			opent.ParamChangefeedID:            changefeedIDStr,
			opent.ParamCDCOperationClusterType: clusterType,
		},
		Actions: []opsvcins.CreateInstanceActionParam{
			{
				Name:        cdcaction.ActionResumeChangefeed.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
		},
	}

	insID, err := c.opSvc.CreateInstance(ctx, createOPparam)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	resp := &ccapi.OperationResp{InstanceID: insID}
	ctrl.Success(ctx, resp)
}

// ShutdownTiCDCCluster
//
//	@Id				CloudConfigControllerShutdownTiCDCCluster
//	@Description	shutdown ticdc cluster
//	@Tags			CloudConfigController
//	@Accept			json
//	@Produce		json
//	@Param			ticdc_cluster_id	path		string					true	"ticdc cluster id"
//	@Param			req					body		ccapi.ShutdownTiCDCReq	true	"request param"
//	@Success		200					{object}	ccapi.OperationResp
//	@Failure		default				{object}	api.BaseResp
//	@Router			/api/v1/devops/ticdc-meta/internal-ticdc/{ticdc_cluster_id}/shutdown [post]
func (c *CloudConfigController) ShutdownTiCDCCluster(ctx *gin.Context) {
	ticdcClusterIDStr := ctx.Param("ticdc_cluster_id")
	if ticdcClusterIDStr == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("ticdc_cluster_id is required"), api.ErrCodeInvalidArgument)
		return
	}
	ticdcClusterID, err := strconv.ParseUint(ticdcClusterIDStr, 10, 64)
	if err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	req := &ccapi.ShutdownTiCDCReq{}
	if err := ctx.ShouldBindJSON(req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	user, ok := ctx.Get(consts.CtxUserProfile)
	if !ok {
		ctrl.FailWithCode(ctx, fmt.Errorf("user profile not found"), api.ErrCodeUnauthenticated)
		return
	}
	userProfile := user.(*userapiv1.PrincipalProfile)

	cdcInfo, err := c.svc.GetInternalTiCDCCluster(ctx, &dataflow.GetInternalTiCDCClusterRequest{TicdcClusterId: ticdcClusterID})
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	if cdcInfo.ControlplaneArch != CDC3LAYERARCH {
		ctrl.FailWithCode(ctx, fmt.Errorf("only 3layer architecture ticdc cluster is supported to shutdown"), api.ErrCodeInvalidArgument)
		return
	}

	createOPparam := opsvc.CreateInstanceParam{
		Type:      opent.OperationTargetCluster,
		Operation: "ShutdownTiCDC",
		Applicant: userProfile.UserEmail,
		Reason:    "",
		Actions: []opsvcins.CreateInstanceActionParam{
			{
				Name:        cdcaction.ActionTiCDCPauseReconcile.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
			{
				Name:        opsaction.ActionApplyConfigChangeViaConfigCenter.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
			{
				Name:        cdcaction.ActionTiCDCDeletePodsAndResumeReconcile.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
		},
		Params: map[opent.ParamName]string{
			opent.ParamClusterID:      strconv.FormatUint(cdcInfo.TidbClusterId, 10),
			opent.ParamCDCClusterID:   ticdcClusterIDStr,
			opent.ParamConfigChangeID: fmt.Sprintf("%d", req.CloudConfigChangeID),
		},
	}

	insID, err := c.opSvc.CreateInstance(ctx, createOPparam)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	resp := &ccapi.OperationResp{InstanceID: insID}
	ctrl.Success(ctx, resp)

}