package cmd

import (
	"github.com/spf13/cobra"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/container"
	bootstarpSerer "github.com/tidbcloud/dataflow-service/internal/cmd/bootstrap/server"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"go.uber.org/zap"
)

func registerClusterJobFunc(cmd *cobra.Command) {
	{
		var clusterID, changefeedID uint64
		deregisterCDCClusterJob := &cobra.Command{
			Use:   "deregisterChangefeed",
			Short: "deregisterChangefeed",
			RunE: func(cmd *cobra.Command, args []string) error {
				if _, err := bootstarpSerer.NewServer(cmd.Context(), bootstarpSerer.ServerOption{
					EnableEventBusConsumer: false,
				}); err != nil {
					return err
				}
				ctx := log.WithTraceID(cmd.Context())
				app := container.Global()
				bizErr := app.ClusterJobModel.DeregisterCDCClusterJob(ctx, clusterID, changefeedID)
				if bizErr != nil {
					log.GetLogger(ctx).Error("DeregisterCDCClusterJob", zap.Error(bizErr))
					return bizErr
				}
				log.GetLogger(ctx).Info("DeregisterCDCClusterJob success", zap.Any("clusterID", clusterID), zap.Any("changefeedID", changefeedID))
				return nil
			},
		}
		deregisterCDCClusterJob.Flags().Uint64VarP(&changefeedID, "changefeedID", "", 0, "")
		deregisterCDCClusterJob.Flags().Uint64VarP(&clusterID, "clusterID", "", 0, "")
		cmd.AddCommand(deregisterCDCClusterJob)
	}
	{
		var clusterID uint64
		listClusterJobs := &cobra.Command{
			Use:   "listClusterJobs",
			Short: "listClusterJobs",
			RunE: func(cmd *cobra.Command, args []string) error {
				if _, err := bootstarpSerer.NewServer(cmd.Context(), bootstarpSerer.ServerOption{
					EnableEventBusConsumer: false,
				}); err != nil {
					return err
				}
				ctx := log.WithTraceID(cmd.Context())
				app := container.Global()
				listJobReq := &core.ListClusterJobsReq{
					Filter: &core.ClusterJobFilter{
						JobType:    core.ClusterJobType_CLUSTER_JOB_TYPE_CDC,
						ClusterIds: []uint64{clusterID},
					},
				}
				resp, bizErr := app.CoreGrpcClient.ListClusterJobs(ctx, listJobReq)
				if bizErr != nil {
					log.GetLogger(ctx).Error("listClusterJobs", zap.Error(bizErr))
					return bizErr
				}
				log.GetLogger(ctx).Info("listClusterJobs success",
					zap.Any("clusterID", clusterID),
					zap.Any("resp", resp))
				return nil
			},
		}
		listClusterJobs.Flags().Uint64VarP(&clusterID, "clusterID", "", 0, "")
		cmd.AddCommand(listClusterJobs)
	}
}

func init() {
	ClusterJobCmd := &cobra.Command{Use: "cluster_job", Short: "cluster job cmd"}
	ClusterJobCmd.PreRunE = func(cmd *cobra.Command, args []string) error {
		return nil
	}
	registerClusterJobFunc(ClusterJobCmd)
	jobCmd.AddCommand(ClusterJobCmd)
}