package cmd

import (
	"encoding/json"
	"fmt"

	"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"
	bootstarpSerer "github.com/tidbcloud/dataflow-service/internal/cmd/bootstrap/server"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"go.uber.org/zap"
)

// TODO(ideascf): should we keep these jobs? or should we revive them as we change the underlying model?
func registerTiCDCClusterJobFunc(cmd *cobra.Command) {
	{
		var createReqString string
		suBcmd := &cobra.Command{
			Use:   "create-ticdc-cluster",
			Short: "create-ticdc-cluster",
			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()
				createReq := &domain.CreateTiCDCClusterReq{}
				err := json.Unmarshal([]byte(createReqString), createReq)
				if err != nil {
					log.GetLogger(ctx).Error("unmarshal createReqString", zap.Error(err))
					return err
				}
				resp, err := app.RegionalTiCDCClusterModel.Create(ctx, createReq)
				if err != nil {
					log.GetLogger(ctx).Error("create ticdc cluster", zap.Error(err))
					return err
				}
				log.GetLogger(ctx).Info("create ticdc cluster success", zap.Any("resp", resp))
				return nil
			},
		}
		suBcmd.Flags().StringVarP(&createReqString, "request", "", "", "")
		cmd.AddCommand(suBcmd)
	}

	{
		var scaleReqString string
		suBcmd := &cobra.Command{
			Use:   "scale-ticdc-cluster",
			Short: "scale-ticdc-cluster",
			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()
				scaleReq := &domain.ScaleTiCDCClusterReq{}
				err := json.Unmarshal([]byte(scaleReqString), scaleReq)
				if err != nil {
					log.GetLogger(ctx).Error("unmarshal scaleReqString", zap.Error(err))
					return err
				}
				resp, err := app.RegionalTiCDCClusterModel.Scale(ctx, scaleReq)
				if err != nil {
					log.GetLogger(ctx).Error("scale ticdc cluster", zap.Error(err))
					return err
				}
				log.GetLogger(ctx).Info("scale ticdc cluster success", zap.Any("resp", resp))
				return nil
			},
		}
		suBcmd.Flags().StringVarP(&scaleReqString, "request", "", "", "")
		cmd.AddCommand(suBcmd)
	}

	{
		var ticdcClusterID uint64
		suBcmd := &cobra.Command{
			Use:   "delete-ticdc-cluster",
			Short: "delete-ticdc-cluster",
			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()
				resp, err := app.RegionalTiCDCClusterModel.Delete(ctx, &domain.DeleteTiCDCClusterReq{
					ID: ticdcClusterID,
				})
				if err != nil {
					log.GetLogger(ctx).Error("delete ticdc cluster", zap.Error(err))
					return err
				}
				log.GetLogger(ctx).Info("delete ticdc cluster success", zap.Any("resp", resp))
				return nil
			},
		}
		suBcmd.Flags().Uint64VarP(&ticdcClusterID, "ticdcClusterID", "", 0, "")
		cmd.AddCommand(suBcmd)
	}

	{
		var ticdcClusterID uint64
		suBcmd := &cobra.Command{
			Use:   "apply-ticdc-cluster-config",
			Short: "apply-ticdc-cluster-config",
			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()
				err := app.RegionalTiCDCClusterModel.ApplyConfig(ctx, &domain.ApplyTiCDCClusterConfigReq{
					TiCDCClusterID: ticdcClusterID,
				})
				if err != nil {
					log.GetLogger(ctx).Error("apply ticdc cluster config error", zap.Error(err))
					return err
				}
				log.GetLogger(ctx).Info("apply ticdc cluster config success")
				return nil
			},
		}
		suBcmd.Flags().Uint64VarP(&ticdcClusterID, "ticdcClusterID", "", 0, "")
		cmd.AddCommand(suBcmd)
	}

	{
		var createReqString string
		suBcmd := &cobra.Command{
			Use:   "create-changefeed",
			Short: "create-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())
				app := container.Global()
				createReq := &domain.CreateChangefeedReq{}
				err := json.Unmarshal([]byte(createReqString), createReq)
				if err != nil {
					log.GetLogger(ctx).Error("unmarshal createReqString", zap.Error(err))
					return err
				}
				resp, err := app.RegionalTiCDCChangefeedModel.CreateChangefeed(ctx, createReq)
				if err != nil {
					log.GetLogger(ctx).Error("create changefeed cluster", zap.Error(err))
					return err
				}
				log.GetLogger(ctx).Info("create changefeed success", zap.Any("resp", resp))
				return nil
			},
		}
		suBcmd.Flags().StringVarP(&createReqString, "request", "", "", "")
		cmd.AddCommand(suBcmd)
	}

	{
		var modifyReqString string
		suBcmd := &cobra.Command{
			Use:   "modify-changefeed",
			Short: "modify-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())
				app := container.Global()
				modifyReq := &domain.ModifyChangefeedConfigReq{}
				err := json.Unmarshal([]byte(modifyReqString), modifyReq)
				if err != nil {
					log.GetLogger(ctx).Error("unmarshal modifyReqString", zap.Error(err))
					return err
				}
				resp, err := app.RegionalTiCDCChangefeedModel.ModifyChangefeedConfig(ctx, modifyReq)
				if err != nil {
					log.GetLogger(ctx).Error("modify changefeed cluster", zap.Error(err))
					return err
				}
				log.GetLogger(ctx).Info("modify changefeed success", zap.Any("resp", resp))
				return nil
			},
		}
		suBcmd.Flags().StringVarP(&modifyReqString, "request", "", "", "")
		cmd.AddCommand(suBcmd)
	}

	{
		var changefeedID uint64
		suBcmd := &cobra.Command{
			Use:   "delete-changefeed",
			Short: "delete-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())
				app := container.Global()
				resp, err := app.RegionalTiCDCChangefeedModel.DeleteChangefeed(ctx,
					&domain.DeleteChangefeedReq{ChangefeedID: changefeedID})
				if err != nil {
					log.GetLogger(ctx).Error("delete changefeed cluster", zap.Error(err))
					return err
				}
				log.GetLogger(ctx).Info("delete changefeed success", zap.Any("resp", resp))
				return nil
			},
		}
		suBcmd.Flags().Uint64VarP(&changefeedID, "changefeedID", "", 0, "")
		cmd.AddCommand(suBcmd)
	}

	{
		var changefeedID uint64
		suBcmd := &cobra.Command{
			Use:   "mark-changefeed-deleted",
			Short: "mark-changefeed-deleted",
			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()
				_, err := app.RegionalTiCDCChangefeedModel.MarkChangefeedDeleted(ctx, &domain.MarkChangefeedDeletedReq{
					ChangefeedID: changefeedID,
				})
				if err != nil {
					log.GetLogger(ctx).Error("mark changefeed deleted failed", zap.Error(err))
					return err
				}
				log.GetLogger(ctx).Info("mark changefeed deleted success")
				return nil
			},
		}
		suBcmd.Flags().Uint64VarP(&changefeedID, "changefeedID", "", 0, "")
		cmd.AddCommand(suBcmd)
	}

	{
		var changefeedID uint64
		suBcmd := &cobra.Command{
			Use:   "pause-changefeed",
			Short: "pause-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())
				app := container.Global()
				resp, err := app.RegionalTiCDCChangefeedModel.PauseChangefeed(ctx,
					&domain.PauseChangefeedReq{ChangefeedID: changefeedID})
				if err != nil {
					log.GetLogger(ctx).Error("pause changefeed cluster", zap.Error(err))
					return err
				}
				log.GetLogger(ctx).Info("pause changefeed success", zap.Any("resp", resp))
				return nil
			},
		}
		suBcmd.Flags().Uint64VarP(&changefeedID, "changefeedID", "", 0, "")
		cmd.AddCommand(suBcmd)
	}

	{
		var changefeedID uint64
		suBcmd := &cobra.Command{
			Use:   "resume-changefeed",
			Short: "resume-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())
				app := container.Global()
				resp, err := app.RegionalTiCDCChangefeedModel.ResumeChangefeed(ctx,
					&domain.ResumeChangefeedReq{ChangefeedID: changefeedID})
				if err != nil {
					log.GetLogger(ctx).Error("resume changefeed cluster", zap.Error(err))
					return err
				}
				log.GetLogger(ctx).Info("resume changefeed success", zap.Any("resp", resp))
				return nil
			},
		}
		suBcmd.Flags().Uint64VarP(&changefeedID, "changefeedID", "", 0, "")
		cmd.AddCommand(suBcmd)
	}

	{
		suBcmd := &cobra.Command{
			Use:   "sync-all-changefeed",
			Short: "sync-all-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())
				app := container.Global()
				err := app.RegionalTiCDCChangefeedWorker.SyncAllChangefeed(ctx)
				if err != nil {
					log.GetLogger(ctx).Error("sync changefeed error", zap.Error(err))
					return err
				}
				log.GetLogger(ctx).Info("sync changefeed success")
				return nil
			},
		}
		cmd.AddCommand(suBcmd)
	}

	{
		var region string
		subCmd := &cobra.Command{
			Use:   "compensate-cloud-resource-tags",
			Short: "compensate-cloud-resource-tags",
			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()
				op := func(qs entity.TiCDCClusterQuerySet) entity.TiCDCClusterQuerySet {
					qs = qs.StatusNotIn(entity.TiCDCClusterStatusDeleting, entity.TiCDCClusterStatusDeleted).
						IsRegionalEq(true).
						InfraProviderTypeIn(
							string(domain.InfraProviderTypeEKS), string(domain.InfraProviderTypeAWS))
					if region != "" {
						qs = qs.RegionEq(region)
					}
					return qs
				}
				cdcClusters, err := app.TiCDCClusterRepo.List(ctx, op)
				if err != nil {
					log.GetLogger(ctx).Error("list ticdc clusters error", zap.Error(err))
					return err
				}
				tenantRegions := make(map[string]*regionalapi.Region)
				for _, cdcCluster := range cdcClusters {
					k := fmt.Sprintf("%d-%s", cdcCluster.TenantID, cdcCluster.Region)
					if _, ok := tenantRegions[k]; !ok {
						tenantRegions[k] = &regionalapi.Region{
							TenantID:      cdcCluster.TenantID.Uint64(),
							Name:          cdcCluster.Region,
							CloudProvider: string(domain.CloudProviderTypeAWS),
						}
					}
				}

				var failedRegions []*regionalapi.Region
				for _, region := range tenantRegions {
					logger := log.GetLogger(ctx).With(
						zap.Uint64("tenant_id", region.TenantID), zap.String("region", region.Name))
					if err := app.TiCDCClusterAPIClient.CompensateComputeCloudResourceTags(
						ctx, region, &ticdc_v1.CompensateComputeCloudResourceTagsRequest{}); err != nil {
						failedRegions = append(failedRegions, region)
						logger.Error("compensate cloud resource tags error", zap.Error(err))
						continue
					}
					logger.Info("compensate cloud resource tags successfully")
				}

				if len(failedRegions) > 0 {
					log.GetLogger(ctx).Error("some regions fail to compensate cloud resource labels", zap.Any("regions", failedRegions))
					return fmt.Errorf("some regions fail to compensate cloud resource labels")
				}

				log.GetLogger(ctx).Info("all regions are successful")
				return nil
			},
		}
		subCmd.Flags().StringVarP(&region, "region", "", "", "specify region of ticdc cluster. if it is empty, it means all regions")
		cmd.AddCommand(subCmd)
	}
}

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

	registerTiCDCClusterJobFunc(cmd)
	jobCmd.AddCommand(cmd)
}