package cmd

import (
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/spf13/cobra"
	"github.com/tidbcloud/dataflow-service-ng/app/container"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	bootstarpSerer "github.com/tidbcloud/dataflow-service-ng/internal/cmd/bootstrap/server"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"go.uber.org/zap"
)

func registerBizChangefeedJobFun(cmd *cobra.Command) {
	{
		var (
			clusterID           uint64
			downstreamParamsStr string
			specificationCode   string
		)
		subCmd := &cobra.Command{
			Use:   "create",
			Short: "create a biz_changefeed",
			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())
				logger := log.GetLogger(ctx)
				app := container.Global()

				downstreamParams := new(dataflow.MySQLInfo)
				decoder := json.NewDecoder(strings.NewReader(downstreamParamsStr))
				decoder.DisallowUnknownFields()
				if err := decoder.Decode(downstreamParams); err != nil {
					logger.Error("decode downstream params error", zap.Error(err))
					return err
				}
				name := fmt.Sprintf("biz_changefeed-%d-%d", clusterID, time.Now().Unix())

				req := &domain.CreateBizChangefeedReq{
					Name:                name,
					ClusterID:           clusterID,
					DownstreamType:      dataflow.ChangefeedType_MYSQL,
					DownstreamParameter: downstreamParams,
					SpecificationCode:   specificationCode,
					UserID:              1,
				}
				resp, bizErr := app.BizChangefeedModel.CreateBizChangefeed(ctx, req)
				if bizErr != nil {
					log.WithRedactedRequest(logger, req).Error("create biz_changefeed error", zap.Error(bizErr))
					return bizErr
				}
				logger.Info("create biz_changefeed successfully",
					zap.String("workflow_id", resp.WorkflowID),
					zap.Uint64("biz_changefeed_id", resp.BizChangefeedID))
				return nil
			},
		}
		subCmd.Flags().Uint64VarP(&clusterID, "clusterID", "", 0, "")
		subCmd.Flags().StringVarP(&specificationCode, "specCode", "", "", "")
		subCmd.Flags().StringVarP(&downstreamParamsStr, "params", "", "", "")
		cmd.AddCommand(subCmd)
	}

	{
		var specificationCode string
		subCmd := &cobra.Command{
			Use:   "scale [biz_changefeed_id]",
			Short: "scale a biz_changefeed",
			RunE: func(cmd *cobra.Command, args []string) error {
				if _, err := bootstarpSerer.NewServer(cmd.Context(), bootstarpSerer.ServerOption{
					EnableEventBusConsumer: false,
				}); err != nil {
					return err
				}
				bizChangefeedID, err := strconv.ParseUint(args[0], 10, 64)
				if err != nil {
					return err
				}
				ctx := log.WithTraceID(cmd.Context())
				logger := log.GetLogger(ctx)
				app := container.Global()

				req := &domain.ScaleBizChangefeedReq{
					BizChangefeedID:   bizChangefeedID,
					SpecificationCode: specificationCode,
					UserID:            1,
				}
				resp, bizErr := app.BizChangefeedModel.ScaleBizChangefeed(ctx, req)
				if bizErr != nil {
					logger.Error("scale biz_changefeed error", zap.Any("req", req), zap.Error(bizErr))
					return bizErr
				}
				logger.Info("scale biz_changefeed successfully", zap.String("workflow_id", resp.WorkflowID))
				return nil
			},
		}
		subCmd.Flags().StringVarP(&specificationCode, "specCode", "", "", "")
		cmd.AddCommand(subCmd)
	}

	{
		subCmd := &cobra.Command{
			Use:   "delete [biz_changefeed_id]",
			Short: "delete a biz_changefeed",
			RunE: func(cmd *cobra.Command, args []string) error {
				if _, err := bootstarpSerer.NewServer(cmd.Context(), bootstarpSerer.ServerOption{
					EnableEventBusConsumer: false,
				}); err != nil {
					return err
				}
				bizChangefeedID, err := strconv.ParseUint(args[0], 10, 64)
				if err != nil {
					return err
				}
				ctx := log.WithTraceID(cmd.Context())
				logger := log.GetLogger(ctx)
				app := container.Global()

				req := &domain.DeleteBizChangefeedReq{
					BizChangefeedID: bizChangefeedID,
					UserID:          1,
				}
				resp, bizErr := app.BizChangefeedModel.DeleteBizChangefeed(ctx, req)
				if bizErr != nil {
					logger.Error("delete biz_changefeed error", zap.Any("req", req), zap.Error(bizErr))
					return bizErr
				}
				logger.Info("delete biz_changefeed successfully", zap.String("workflow_id", resp.WorkflowID))
				return nil
			},
		}
		cmd.AddCommand(subCmd)
	}

	{
		subCmd := &cobra.Command{
			Use:   "pause [biz_changefeed_id]",
			Short: "pause a biz_changefeed",
			RunE: func(cmd *cobra.Command, args []string) error {
				if _, err := bootstarpSerer.NewServer(cmd.Context(), bootstarpSerer.ServerOption{
					EnableEventBusConsumer: false,
				}); err != nil {
					return err
				}
				bizChangefeedID, err := strconv.ParseUint(args[0], 10, 64)
				if err != nil {
					return err
				}
				ctx := log.WithTraceID(cmd.Context())
				logger := log.GetLogger(ctx)
				app := container.Global()

				req := &domain.PauseBizChangefeedReq{
					BizChangefeedID: bizChangefeedID,
					UserID:          1,
				}
				bizErr := app.BizChangefeedModel.PauseBizChangefeed(ctx, req)
				if bizErr != nil {
					logger.Error("pause biz_changefeed error", zap.Any("req", req), zap.Error(bizErr))
					return bizErr
				}
				logger.Info("pause biz_changefeed successfully")
				return nil
			},
		}
		cmd.AddCommand(subCmd)
	}

	{
		subCmd := &cobra.Command{
			Use:   "resume [biz_changefeed_id]",
			Short: "resume a biz_changefeed",
			RunE: func(cmd *cobra.Command, args []string) error {
				if _, err := bootstarpSerer.NewServer(cmd.Context(), bootstarpSerer.ServerOption{
					EnableEventBusConsumer: false,
				}); err != nil {
					return err
				}
				bizChangefeedID, err := strconv.ParseUint(args[0], 10, 64)
				if err != nil {
					return err
				}
				ctx := log.WithTraceID(cmd.Context())
				logger := log.GetLogger(ctx)
				app := container.Global()

				req := &domain.ResumeBizChangefeedReq{
					BizChangefeedID: bizChangefeedID,
					UserID:          1,
				}
				bizErr := app.BizChangefeedModel.ResumeBizChangefeed(ctx, req)
				if bizErr != nil {
					logger.Error("resume biz_changefeed error", zap.Any("req", req), zap.Error(bizErr))
					return bizErr
				}
				logger.Info("resume biz_changefeed successfully")
				return nil
			},
		}
		cmd.AddCommand(subCmd)
	}

	{
		var name, downstreamParamsStr string
		subCmd := &cobra.Command{
			Use:   "edit [biz_changefeed_id]",
			Short: "edit a biz_changefeed",
			RunE: func(cmd *cobra.Command, args []string) error {
				if _, err := bootstarpSerer.NewServer(cmd.Context(), bootstarpSerer.ServerOption{
					EnableEventBusConsumer: false,
				}); err != nil {
					return err
				}
				bizChangefeedID, err := strconv.ParseUint(args[0], 10, 64)
				if err != nil {
					return err
				}
				ctx := log.WithTraceID(cmd.Context())
				logger := log.GetLogger(ctx)
				app := container.Global()

				downstreamParams := new(dataflow.MySQLInfo)
				decoder := json.NewDecoder(strings.NewReader(downstreamParamsStr))
				decoder.DisallowUnknownFields()
				if err := decoder.Decode(downstreamParams); err != nil {
					logger.Error("decode downstream params error", zap.Error(err))
					return err
				}

				req := &domain.EditBizChangefeedReq{
					BizChangefeedID:     bizChangefeedID,
					Name:                name,
					DownstreamType:      dataflow.ChangefeedType_MYSQL,
					DownstreamParameter: downstreamParams,
					UserID:              1,
				}
				bizErr := app.BizChangefeedModel.EditBizChangefeed(ctx, req)
				if bizErr != nil {
					log.WithRedactedRequest(logger, req).Error("edit biz_changefeed error", zap.Error(bizErr))
					return bizErr
				}
				logger.Info("edit biz_changefeed successfully")
				return nil
			},
		}
		subCmd.Flags().StringVarP(&name, "name", "", "", "")
		subCmd.Flags().StringVarP(&downstreamParamsStr, "params", "", "", "")
		cmd.AddCommand(subCmd)
	}

	{
		var (
			clusterID           uint64
			downstreamParamsStr string
		)
		subCmd := &cobra.Command{
			Use:   "test",
			Short: "test a biz_changefeed",
			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())
				logger := log.GetLogger(ctx)
				app := container.Global()

				downstreamParams := new(dataflow.MySQLInfo)
				decoder := json.NewDecoder(strings.NewReader(downstreamParamsStr))
				decoder.DisallowUnknownFields()
				if err := decoder.Decode(downstreamParams); err != nil {
					logger.Error("decode downstream params error", zap.Error(err))
					return err
				}

				req := &domain.TestBizChangefeedReq{
					ClusterID:           clusterID,
					DownstreamType:      dataflow.ChangefeedType_MYSQL,
					DownstreamParameter: downstreamParams,
				}
				bizErr := app.BizChangefeedModel.TestBizChangefeed(ctx, req)
				if bizErr != nil {
					log.WithRedactedRequest(logger, req).Error("test biz_changefeed error", zap.Error(bizErr))
					return bizErr
				}
				logger.Info("test biz_changefeed successfully")
				return nil
			},
		}
		subCmd.Flags().Uint64VarP(&clusterID, "clusterID", "", 0, "")
		subCmd.Flags().StringVarP(&downstreamParamsStr, "params", "", "", "")
		cmd.AddCommand(subCmd)
	}
}

func init() {
	bizChangefeedCmd := &cobra.Command{
		Use:   "biz_changefeed",
		Short: "biz_changefeed cmd",
		PreRunE: func(cmd *cobra.Command, args []string) error {
			return nil
		},
	}

	registerBizChangefeedJobFun(bizChangefeedCmd)
	jobCmd.AddCommand(bizChangefeedCmd)
}