package service

import (
	"context"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	appworkflow "github.com/tidbcloud/serverless-service/internal/workflow"
)

func (s *cdcService) submitCreateNewArchChangefeedDag(
	_ context.Context,
	changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	if s.workflow == nil || changefeed == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "cdc workflow is not initialized")
	}
	_, err := s.workflow.RunDagWithTags(
		appworkflow.DagCreateNewArchChangefeed,
		map[string]string{
			"cluster_id":    changefeed.ClusterID,
			"changefeed_id": changefeed.ID,
		},
		map[string]string{
			appworkflow.TagWorkflowCategory: appworkflow.DagCreateNewArchChangefeed,
			appworkflow.TagClusterID:        changefeed.ClusterID,
			appworkflow.TagChangefeedID:     changefeed.ID,
			appworkflow.TagProvider:         changefeed.CloudProvider,
			appworkflow.TagRegion:           changefeed.Region,
			appworkflow.TagKubernetes:       changefeed.Kubernetes,
		},
	)
	if err != nil {
		return bizerr.WrapError(err, "submit create newarch changefeed dag failed")
	}
	return nil
}

func (s *cdcService) submitScaleNewArchClusterDag(
	_ context.Context,
	changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	if s.workflow == nil || changefeed == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "cdc workflow is not initialized")
	}
	_, err := s.workflow.RunDagWithTags(
		appworkflow.DagScaleNewArchCluster,
		map[string]string{
			"cluster_id":    changefeed.ClusterID,
			"changefeed_id": changefeed.ID,
		},
		map[string]string{
			appworkflow.TagWorkflowCategory: appworkflow.DagScaleNewArchCluster,
			appworkflow.TagClusterID:        changefeed.ClusterID,
			appworkflow.TagChangefeedID:     changefeed.ID,
			appworkflow.TagProvider:         changefeed.CloudProvider,
			appworkflow.TagRegion:           changefeed.Region,
			appworkflow.TagKubernetes:       changefeed.Kubernetes,
		},
	)
	if err != nil {
		return bizerr.WrapError(err, "submit scale newarch cluster dag failed")
	}
	return nil
}

func (s *cdcService) submitDeleteNewArchChangefeedDag(
	_ context.Context,
	changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	if s.workflow == nil || changefeed == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "cdc workflow is not initialized")
	}
	_, err := s.workflow.RunDagWithTags(
		appworkflow.DagDeleteNewArchChangefeed,
		map[string]string{
			"cluster_id":    changefeed.ClusterID,
			"changefeed_id": changefeed.ID,
		},
		map[string]string{
			appworkflow.TagWorkflowCategory: appworkflow.DagDeleteNewArchChangefeed,
			appworkflow.TagClusterID:        changefeed.ClusterID,
			appworkflow.TagChangefeedID:     changefeed.ID,
			appworkflow.TagProvider:         changefeed.CloudProvider,
			appworkflow.TagRegion:           changefeed.Region,
			appworkflow.TagKubernetes:       changefeed.Kubernetes,
		},
	)
	if err != nil {
		return bizerr.WrapError(err, "submit delete newarch changefeed dag failed")
	}
	return nil
}