package cmd

import (
	"fmt"
	"github.com/spf13/cobra"
	"github.com/tidbcloud/account/idl/pbgen/proto/account"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"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/app/workflow"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	bootstarpSerer "github.com/tidbcloud/dataflow-service/internal/cmd/bootstrap/server"
	"github.com/tidbcloud/dataflow-service/internal/jobs"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
	"strconv"
	"time"
)

const projectStatusActive = "active"

func registerBackupJobFunc(cmd *cobra.Command) {
	{
		var clusterID uint64
		suBcmd := &cobra.Command{
			Use:   "wait_next_time_to_enable_pitr",
			Short: "wait next snapshot backup to enable pitr because of snapshot backup failed",
			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())
				return jobs.WaitNextTimeToEnablePitr(ctx, clusterID)
			},
		}
		suBcmd.Flags().Uint64VarP(&clusterID, "clusterID", "", 0, "")
		cmd.AddCommand(suBcmd)
	}
	{
		var backupID uint64
		suBcmd := &cobra.Command{
			Use:   "deregister_backup_cluster_job",
			Short: "deregister backup cluster job",
			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()
				backup, err := app.BackupRepo.Load(ctx, backupID)
				if err != nil {
					logger.Error("query snapshot backup error", zap.Error(err))
					return err
				}
				return app.ClusterJobModel.DeregisterBackupClusterJob(ctx,
					backup.ClusterID.Uint64(), backup.ID.Uint64())
			},
		}
		suBcmd.Flags().Uint64VarP(&backupID, "backupID", "", 0, "")
		cmd.AddCommand(suBcmd)
	}

	{
		var backupID uint64
		suBcmd := &cobra.Command{
			Use:   "delete_snapshot_backup",
			Short: "delete snapshot backup",
			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()
				resp, err := app.SnapshotBackupModel.AsyncDelete(ctx, &domain.AsyncDeleteBackupReq{
					BackupID: backupID,
				})
				if err != nil {
					logger.Error("async delete snapshot backup error", zap.Error(err))
					return err
				}
				logger.Info("async delete snapshot backup success", zap.Any("resp", resp))
				return nil
			},
		}
		suBcmd.Flags().Uint64VarP(&backupID, "backupID", "", 0, "")
		cmd.AddCommand(suBcmd)
	}

	{
		var logBackupID uint64
		suBcmd := &cobra.Command{
			Use:   "delete_log_backup",
			Short: "delete log backup",
			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()
				resp, err := app.PitBackupModel.AsyncDeleteLogBackup(ctx, &domain.DeleteLogBackupReq{
					LogBackupID: logBackupID,
				})
				if err != nil {
					logger.Error("async delete log backup error", zap.Error(err))
					return err
				}
				logger.Info("async delete log backup success", zap.Any("resp", resp))
				return nil
			},
		}
		suBcmd.Flags().Uint64VarP(&logBackupID, "logBackupID", "", 0, "")
		cmd.AddCommand(suBcmd)
	}

	{
		var backupID uint64
		suBcmd := &cobra.Command{
			Use:   "clean_backup_cloud_storage",
			Short: "clean backup cloud storage",
			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()
				backupEntity, err := app.BackupRepo.Load(ctx, backupID)
				if err != nil {
					return err
				}
				vars := map[string]string{workflow.KeyBackupID: strconv.FormatUint(uint64(backupEntity.ID), 10)}
				tags := map[string]string{
					workflow.BusinessID:        strconv.FormatUint(uint64(backupEntity.ID), 10),
					workflow.BusinessTypeKey:   string(workflow.BusinessTypeSnapshotBackup),
					workflow.BusinessActionKey: string(workflow.BusinessActionSnapshotBackupCleanStorage),
				}
				res, rErr := app.WorkflowModel.RunDagWithTags(workflow.DagCleanupSnapshotBackupStorage, vars, tags)
				if rErr != nil {
					logger.Error("clean failed backup storage start dag failed", zap.Error(rErr))
					return rErr
				}
				logger.Info("clean failed backup storage dag successfully", zap.String("dagInsID", res.ID))
				return nil
			},
		}
		suBcmd.Flags().Uint64VarP(&backupID, "backupID", "", 0, "")
		cmd.AddCommand(suBcmd)
	}

	{
		var (
			clusterID   uint64
			startTS     int64
			tableFilter []string
		)
		subCmd := &cobra.Command{
			Use:   "create_snapshot_backup",
			Short: "create snapshot backup",
			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()
				var expectedBackupTS string
				if startTS > 0 {
					expectedBackupTS = util.TransToTSOStr(startTS)
				}
				req := &domain.AsyncCreateBackupReq{
					Name:             fmt.Sprintf("munual-test-%d", time.Now().Unix()),
					ClusterID:        clusterID,
					BackupType:       dataflow.BackupType_MANUAL,
					TableFilter:      tableFilter,
					ExpectedBackupTS: expectedBackupTS,
				}
				resp, err := app.SnapshotBackupModel.AsyncCreate(ctx, req)
				if err != nil {
					logger.Error("create backup error", zap.Error(err), zap.Any("req", req))
					return err
				} else {
					logger.Info("create backup successfully", zap.Any("req", req), zap.Any("resp", resp))
					return nil
				}
			},
		}
		subCmd.Flags().Uint64VarP(&clusterID, "clusterID", "", 0, "")
		subCmd.Flags().Int64VarP(&startTS, "startTS", "", 0, "")
		subCmd.Flags().StringSliceVarP(&tableFilter, "tableFilter", "", nil, "")
		cmd.AddCommand(subCmd)
	}

	{
		subCmd := &cobra.Command{
			Use:   "clean_unused_backup_schedules",
			Short: "clean unused backup schedules",
			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()

				backupSchedules, err := app.BackupScheduleRepo.List(ctx,
					func(qs entity.BackupScheduleQuerySet) entity.BackupScheduleQuerySet {
						return qs.ClusterTypeEq(entity.BackupClusterTypeDedicated).
							DeletedAtIsNull().
							TypeEq(entity.BackupTypeAuto)
					})
				if err != nil {
					logger.Error("list backup schedules error", zap.Error(err))
					return err
				}
				for _, backupSchedule := range backupSchedules {
					logger := logger.With(zap.Uint64("cluster_id", backupSchedule.ClusterID.Uint64()),
						zap.Uint64("backup_schedule_id", backupSchedule.ID.Uint64()))
					cluster, err := app.CoreGrpcClient.GetAnyClusterByID(ctx, backupSchedule.ClusterID.Uint64())
					if err == nil {
						if cluster.Status == core.ClusterStatus_CLUSTER_STATUS_DELETED ||
							cluster.Status == core.ClusterStatus_CLUSTER_STATUS_DELETECOMPLETE ||
							cluster.Status == core.ClusterStatus_CLUSTER_STATUS_CLEARED {
							logger.Info("cluster is deleted, try to delete the backup schedule",
								zap.String("cluster_status", cluster.Status.String()))
							// delete backup schedule
							if err := app.BackupScheduleRepo.SoftDelete(ctx, backupSchedule.ID); err != nil {
								logger.Warn("delete backup schedule error", zap.Error(err))
							} else {
								logger.Info("delete unused auto backup scheduler successfully")
							}
						}
						continue
					} else if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
						logger.Warn("get cluster error", zap.Error(err))
						continue
					}

					// cluster is not found, check if the project is deleted
					backup, err := app.BackupRepo.Get(ctx, func(qs entity.BackupQuerySet) entity.BackupQuerySet {
						return qs.BackupScheduleIDEq(backupSchedule.ID).OrderDescByCreatedAt()
					})
					if err != nil {
						logger.Warn("get latest backup error", zap.Error(err))
						continue
					}
					projectRes, err := app.AccountSvcGrpcClient.GetProjectByID(ctx, &account.GetProjectByIDRequest{
						Id: backup.ProjectID.Uint64(),
					})
					if err != nil {
						logger.Warn("get project error",
							zap.Uint64("project_id", backup.ProjectID.Uint64()), zap.Error(err))
						continue
					}
					if projectRes.Project.Status == projectStatusActive {
						logger.Warn("cluster is not found, but project is active",
							zap.Uint64("project_id", backup.ProjectID.Uint64()))
						continue
					}
					logger.Info("project is not active, try to delete the backup schedule",
						zap.Uint64("project_id", backup.ProjectID.Uint64()),
						zap.String("project_status", projectRes.Project.Status))
					// delete backup schedule
					if err := app.BackupScheduleRepo.SoftDelete(ctx, backupSchedule.ID); err != nil {
						logger.Warn("delete backup schedule error", zap.Error(err))
					} else {
						logger.Info("delete unused auto backup scheduler successfully")
					}
				}
				return nil
			},
		}
		cmd.AddCommand(subCmd)
	}
}

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

	registerBackupJobFunc(backupCmd)
	jobCmd.AddCommand(backupCmd)
}