package cmd

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

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

func registerBizChangefeedJobFun(cmd *cobra.Command) {
	{
		var (
			clusterID             uint64
			downstreamParamsStr   string
			specificationCode     string
			downstreamType        int32
			networkInfoStr        string
			privateLinkEndpointID uint64
			deployType            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()

				connectorType := dataflow.ConnectorType(downstreamType)

				var downstreamParams interface{}
				switch connectorType {
				case dataflow.ConnectorType_KAFKA:
					downstreamParams = new(dataflow.KafkaInfo)
				case dataflow.ConnectorType_MYSQL:
					downstreamParams = new(dataflow.MySQLInfo)
				case dataflow.ConnectorType_S3:
					downstreamParams = new(dataflow.S3Info)
				case dataflow.ConnectorType_GCS:
					downstreamParams = new(dataflow.GcsInfo)
				case dataflow.ConnectorType_AZURE_BLOB_STORAGE:
					downstreamParams = new(dataflow.AzureBlobStorageInfo)
				case dataflow.ConnectorType_SERVERLESS_TIDB:
					downstreamParams = new(dataflow.ServerlessTidbInfo)
				case dataflow.ConnectorType_PULSAR:
					downstreamParams = new(dataflow.PulsarInfo)
				default:
					return fmt.Errorf("invalid connector type: %s", connectorType)
				}

				decoder := json.NewDecoder(strings.NewReader(downstreamParamsStr))
				decoder.DisallowUnknownFields()
				if err := decoder.Decode(downstreamParams); err != nil {
					return fmt.Errorf("decode downstream params error: %w", err)
				}

				networkInfo := new(dataflow.NetworkInfo)
				if networkInfoStr != "" {
					if err := json.Unmarshal([]byte(networkInfoStr), networkInfo); err != nil {
						return fmt.Errorf("unmarshal network info error: %w", err)
					}
				}

				name := fmt.Sprintf("biz_changefeed-%d-%d", clusterID, time.Now().Unix())
				req := &domain.CreateBizChangefeedReq{
					Name:                  name,
					ClusterID:             clusterID,
					DownstreamType:        connectorType,
					DownstreamParameter:   downstreamParams,
					SpecificationCode:     specificationCode,
					UserID:                1,
					NetworkInfo:           networkInfo,
					DeployType:            entity.TiCDCClusterDeployType(deployType),
					PrivateLinkEndpointID: entity.ID(privateLinkEndpointID),
				}
				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", "", "", "")
		subCmd.Flags().Int32VarP(&downstreamType, "downstreamType", "", 0, "")
		subCmd.Flags().StringVarP(&networkInfoStr, "networkInfo", "", "", "")
		subCmd.Flags().StringVarP(&deployType, "deployType", "", "shared", "")
		subCmd.Flags().Uint64VarP(&privateLinkEndpointID, "privateLinkEndpointID", "", 0, "")
		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.ConnectorType_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.ConnectorType_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)
	}

	{
		subCmd := &cobra.Command{
			Use:   "backfillAllConfig",
			Short: "backfill config of all biz_changefeeds",
			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()

				filterOp := func(qs entity.BizChangefeedQuerySet) entity.BizChangefeedQuerySet {
					return qs.OperationStatusNotIn(
						entity.BizChangefeedOperationStatusDeleting, entity.BizChangefeedOperationStatusDeleted)
				}
				bizChangefeedEnts, err := app.BizChangefeedRepo.List(ctx, filterOp)
				if err != nil {
					logger.Error("list biz_changefeed error", zap.Error(err))
					return err
				}
				logger.Info("list biz_changefeeds successfully", zap.Int("count", len(bizChangefeedEnts)))
				for i := range bizChangefeedEnts {
					bizChangefeedID := bizChangefeedEnts[i].ID.Uint64()
					if !bizChangefeedEnts[i].IsRegional {
						logger.Info(
							"biz_changefeed is not regional", zap.Uint64("biz_changefeed_id", bizChangefeedID))
						continue
					}
					if err := app.BizChangefeedModel.BackfillChangefeedConfig(ctx, bizChangefeedID); err != nil {
						logger.Warn(
							"backfill changefeed config error",
							zap.Uint64("biz_changefeed_id", bizChangefeedID), zap.Error(err))
					}
				}
				logger.Info("backfill config of all biz_changefeed finished")
				return nil
			},
		}
		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)
}