package impl

import (
	"context"
	"fmt"
	"strconv"

	"github.com/Masterminds/semver/v3"
	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/internal/cluster_job"
	"github.com/tidbcloud/dataflow-service/app/internal/status_matrix"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

var _ backup.PitBackupSwitcher = &pitBackupSwitcherImpl{}

type pitBackupSwitcherImpl struct {
	// db
	dbProvider          db.TransProvider
	logBackupRepo       backupRepo.LogBackupRepo
	pitBackupSwitchRepo backupRepo.PitBackupSwitchRepo
	clusterJob          cluster_job.Model

	// internal helpers
	workflowCli  workflow.Workflow
	mgmtCli      grpcli.MgmtSvcGrpcClient
	statusMatrix status_matrix.StatusMatrix
}

func NewPitBackupSwitcher(
	dbProvider db.TransProvider,
	pitBackupSwitchRepo backupRepo.PitBackupSwitchRepo,
	logBackupRepo backupRepo.LogBackupRepo,
	workflowCli workflow.Workflow,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	clusterJob cluster_job.Model,
) backup.PitBackupSwitcher {
	return &pitBackupSwitcherImpl{
		dbProvider:          dbProvider,
		logBackupRepo:       logBackupRepo,
		pitBackupSwitchRepo: pitBackupSwitchRepo,
		clusterJob:          clusterJob,
		workflowCli:         workflowCli,
		mgmtCli:             mgmtCli,
		statusMatrix:        NewPitBackupSwitchStatusMatrix(),
	}
}

func (b *pitBackupSwitcherImpl) Enable(ctx context.Context, req *domain.EnablePitBackupReq) (
	*domain.EnablePitBackupResp, bizerr.BizError,
) {
	logger := log.GetLogger(ctx).Named("pitBackupSwitcherImpl.EnableLogBackup").With(
		zap.Any("cluster id", req.ClusterID))
	ctxWithLogger := log.WithLogger(ctx, logger)
	pitBackupSwitch, err := b.getOrCreatePitBackupSwitch(ctxWithLogger, req.ClusterID)
	if err != nil {
		return nil, err
	}
	cluster, err := b.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, req.ClusterID)
	if err != nil {
		log.GetLogger(ctx).Error("get cluster error", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	if err := b.preCheckOfEnablePitBackup(ctxWithLogger, cluster, pitBackupSwitch); err != nil {
		return nil, err
	}

	if req.DryRun {
		logger.Info("dry run enable pit backup success")
		return nil, nil
	}

	resp := &domain.EnablePitBackupResp{}
	if err := b.dbProvider.Transaction(ctxWithLogger, func(txCtx context.Context) error {
		txCtxWithLogger := log.WithLogger(txCtx, logger)
		nextStatus := b.statusMatrix.CalculateNextStatusForAction(status_matrix.CurrentStatus{
			Status:  string(pitBackupSwitch.Status),
			Actions: pitBackupSwitch.Action.CurrentActions,
		}, string(entity.PitBackupSwitchActionEnable))
		pitBackupSwitch.Action.CurrentActions = nextStatus.Actions
		pitBackupSwitch.Status = entity.PitBackupSwitchStatus(nextStatus.Status)
		// A PiTR restart can replace the log-backup ID. The switch turn is stable
		// for the lifecycle and therefore identifies this pause lock.
		pitBackupSwitch.Turn++
		if err := b.pitBackupSwitchRepo.Save(txCtxWithLogger, pitBackupSwitch); err != nil {
			return err
		}

		resp.PitBackupSwitchID = pitBackupSwitch.ID.Uint64()
		return nil
	}); err != nil {
		logger.Error("enable pit backup switch error", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}

	if err := b.clusterJob.RegisterPiTRPauseLock(ctx, pitBackupSwitch.ClusterID.Uint64(), pitBackupSwitch.Turn); err != nil {
		logger.Error("register pitr pause lock failed", zap.Error(err))
		// Ignore PauseLock registration failure, since the PiTR pause lock is best-effort
	}

	return resp, nil
}

func (b *pitBackupSwitcherImpl) Disable(
	ctx context.Context, req *domain.DisablePitBackupReq,
) (*domain.DisablePitBackupResp, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("pitBackupSwitcherImpl.Disable").With(
		zap.Any("cluster id", req.ClusterID))
	ctxWithLogger := log.WithLogger(ctx, logger)
	pitBackupSwitch, err := b.getOrCreatePitBackupSwitch(ctxWithLogger, req.ClusterID)
	if err != nil {
		return nil, err
	}
	cluster, err := b.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, req.ClusterID)
	if err != nil {
		log.GetLogger(ctx).Error("get cluster error", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	if err := b.preCheckOfDisablePitBackup(ctxWithLogger, cluster, pitBackupSwitch); err != nil {
		return nil, err
	}

	if req.DryRun {
		logger.Info("dry run disable pit backup success")
		return nil, nil
	}

	var resp *domain.DisablePitBackupResp
	transErr := b.dbProvider.Transaction(ctxWithLogger, func(txCtx context.Context) error {
		txCtxWithLogger := log.WithLogger(txCtx, logger)
		var err error
		resp, err = b.executeDisableDag(txCtxWithLogger, logger, pitBackupSwitch, cluster)
		return err
	})
	if transErr != nil {
		log.GetLogger(ctx).Error("disable pit backup failed", zap.Error(transErr))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	return resp, nil
}

func (b *pitBackupSwitcherImpl) executeDisableDag(
	ctx context.Context, logger *zap.Logger, pitBackupSwitch *entity.PitBackupSwitch, cluster *mgmt.Cluster,
) (*domain.DisablePitBackupResp, error) {
	currentStatus := pitBackupSwitch.Status
	nextStatus := b.statusMatrix.CalculateNextStatusForAction(status_matrix.CurrentStatus{
		Status:  string(pitBackupSwitch.Status),
		Actions: pitBackupSwitch.Action.CurrentActions,
	}, string(entity.PitBackupSwitchActionDisable))
	if nextStatus == nil {
		logger.Warn("next status is nil")
		return nil, fmt.Errorf("next status is nil")
	}
	pitBackupSwitch.Status = entity.PitBackupSwitchStatus(nextStatus.Status)
	pitBackupSwitch.Action.CurrentActions = nextStatus.Actions
	switch currentStatus {
	case entity.PitBackupStatusEnabled, entity.PitBackupStatusWaitingForRestarting:
		return b.executeDisableWhenStatusIsEnable(ctx, logger, pitBackupSwitch, cluster)
	case entity.PitBackupStatusWaitingForEnabling:
		if err := b.clusterJob.DeregisterPiTRPauseLock(ctx, pitBackupSwitch.ClusterID.Uint64(), pitBackupSwitch.Turn); err != nil {
			logger.Error("disable pit backup deregister pitr pause lock error", zap.Error(err))
			return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
		}
		return b.executeDisableWhenStatusInWaitingForEnabling(ctx, logger, pitBackupSwitch)
	default:
		return nil, fmt.Errorf("unknown status %s", currentStatus)
	}
}

func (b *pitBackupSwitcherImpl) executeDisableWhenStatusInWaitingForEnabling(
	ctx context.Context, logger *zap.Logger, pitBackupSwitch *entity.PitBackupSwitch,
) (*domain.DisablePitBackupResp, error) {
	pitBackupSwitch.Action.CurrentActions = pie.Of(pitBackupSwitch.Action.CurrentActions).
		Filter(func(s string) bool {
			return s != string(entity.PitBackupSwitchActionDisable)
		}).Result
	if err := b.pitBackupSwitchRepo.Save(ctx, pitBackupSwitch); err != nil {
		logger.Error("disable pit backup save switch error", zap.Error(err))
		return nil, err
	}
	return &domain.DisablePitBackupResp{
		PitBackupSwitchID: pitBackupSwitch.ID.Uint64(),
	}, nil
}

func (b *pitBackupSwitcherImpl) executeDisableWhenStatusIsEnable(
	ctx context.Context, logger *zap.Logger, pitBackupSwitch *entity.PitBackupSwitch, cluster *mgmt.Cluster,
) (*domain.DisablePitBackupResp, error) {
	// get active log backup to run disable pit backup dag
	logBackup, getErr := b.getActiveLogBackup(ctx, cluster)
	if getErr != nil {
		log.GetLogger(ctx).Error("get active log backup error", zap.Error(getErr))
		return nil, getErr
	}
	var dagID string
	dagID, err := b.runDisablePitBackupDag(ctx, logBackup, pitBackupSwitch)
	if err != nil {
		log.GetLogger(ctx).Error("run disable pit backup dag failed", zap.Error(err))
		return nil, err
	}
	if err := b.pitBackupSwitchRepo.Save(ctx, pitBackupSwitch); err != nil {
		logger.Error("disable pit backup save switch error", zap.Error(err))
		return nil, err
	}
	return &domain.DisablePitBackupResp{
		WorkflowID:        dagID,
		PitBackupSwitchID: pitBackupSwitch.ID.Uint64(),
	}, nil
}

func (b *pitBackupSwitcherImpl) preCheckOfEnablePitBackup(
	ctx context.Context, cluster *mgmt.Cluster, pitBackupSwitch *entity.PitBackupSwitch,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	// 1. check the status
	isFriendly := b.statusMatrix.IsActionFriendly2Status(status_matrix.CurrentStatus{
		Status:  string(pitBackupSwitch.Status),
		Actions: pitBackupSwitch.Action.CurrentActions,
	}, string(entity.PitBackupSwitchActionEnable))
	if !isFriendly {
		logger.Warn("modify pit backup switch failed, status is not friendly",
			zap.Any("pit backup switch", pitBackupSwitch))
		return bizerr_wrapper.NewInvalidInputError(
			config.ErrActionParam, pitBackupSwitch.Status, entity.PitBackupSwitchActionEnable)
	}
	// 2. check cluster version
	dc := cluster.GetDedicatedCluster()
	v, err := semver.NewVersion(dc.GetVersion())
	if err != nil {
		logger.Error("parse cluster version error", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	backupCfg := internal_config.Global().BackupConfig
	if !backupCfg.LogBackupEnableMinVerConstraint.Check(v) {
		return bizerr.New(
			config.ErrCodePitBackupPreCheckClusterVersion,
			config.ErrMsgPitBackupPreCheckClusterVersion,
			cluster.GetClusterId(), dc.GetVersion(), backupCfg.LogBackupEnableMinVer)
	}

	// 3. check the cluster status
	if dc.GetState() != dedicatedv1.Cluster_AVAILABLE {
		return bizerr.New(
			config.ErrCodePitBackupPreCheckClusterStatus,
			config.ErrMsgPitBackupPreCheckClusterStatus,
			cluster.GetClusterId(), dc.GetState())
	}
	return nil
}

func (b *pitBackupSwitcherImpl) preCheckOfDisablePitBackup(
	ctx context.Context, cluster *mgmt.Cluster, pitBackupSwitch *entity.PitBackupSwitch,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	isFriendly := b.statusMatrix.IsActionFriendly2Status(status_matrix.CurrentStatus{
		Status:  string(pitBackupSwitch.Status),
		Actions: pitBackupSwitch.Action.CurrentActions,
	}, string(entity.PitBackupSwitchActionDisable))
	if !isFriendly {
		logger.Warn("modify pit backup switch failed, status is not friendly",
			zap.Any("pit backup switch", pitBackupSwitch))
		return bizerr_wrapper.NewInvalidInputError(
			config.ErrActionParam, pitBackupSwitch.Status, entity.PitBackupSwitchActionDisable)
	}
	return nil
}

func (b *pitBackupSwitcherImpl) getOrCreatePitBackupSwitch(
	ctx context.Context, clusterID uint64,
) (*entity.PitBackupSwitch, bizerr.BizError) {
	filterF := func(qs entity.PitBackupSwitchQuerySet) entity.PitBackupSwitchQuerySet {
		return qs.ClusterIDEq(entity.ID(clusterID))
	}
	entList, err := b.pitBackupSwitchRepo.List(ctx, filterF)
	if err != nil {
		log.GetLogger(ctx).Error("list pit backup switch", zap.Error(err))
		return nil, err
	}
	if len(entList) == 0 {
		cluster, err := b.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID)
		if err != nil {
			log.GetLogger(ctx).Error("get cluster error", zap.Error(err))
			return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, clusterID)
		}

		ent := &entity.PitBackupSwitch{
			Status:    entity.PitBackupStatusDisabled,
			ClusterID: entity.ID(cluster.GetClusterId()),
			TenantID:  entity.ID(cluster.GetOrganizationId()),
			ProjectID: entity.ID(cluster.GetProjectId()),
		}
		if err := b.pitBackupSwitchRepo.Save(ctx, ent); err != nil {
			log.GetLogger(ctx).Error("save pit backup switch error", zap.Error(err))
			return nil, err
		}
		return ent, nil
	}
	ent := entList[0]
	return &ent, nil
}

func (b *pitBackupSwitcherImpl) runDisablePitBackupDag(
	ctx context.Context, logBackup *entity.LogBackup, pitBackupSwitch *entity.PitBackupSwitch,
) (string, error) {
	tags := map[string]string{
		workflow.BusinessID:        strconv.FormatUint(pitBackupSwitch.ID.Uint64(), 10),
		workflow.BusinessTypeKey:   string(workflow.BusinessTypePitBackup),
		workflow.BusinessActionKey: string(workflow.BusinessActionDisablePitBackup),
		workflow.ClusterID:         strconv.FormatUint(logBackup.ClusterID.Uint64(), 10),
	}
	vars := map[string]string{
		workflow.KeyPitBackupSwitchID:  strconv.FormatUint(pitBackupSwitch.ID.Uint64(), 10),
		workflow.KeyDisableLogBackupID: strconv.FormatUint(logBackup.ID.Uint64(), 10),
	}
	res, err := b.workflowCli.RunDagWithTags(workflow.DagDisablePitBackup, vars, tags)
	if err != nil {
		log.GetLogger(ctx).Error("create dag failed", zap.Error(err))
		return "", err
	}
	return res.ID, nil
}

//nolint:dupl
func (b *pitBackupSwitcherImpl) getActiveLogBackup(
	ctx context.Context, cluster *mgmt.Cluster,
) (*entity.LogBackup, bizerr.BizError) {
	filterF := func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		qs = qs.ClusterIDEq(entity.ID(cluster.GetClusterId())).
			ProjectIDEq(entity.ID(cluster.GetProjectId())).
			TenantIDEq(entity.ID(cluster.GetOrganizationId())).
			DeletedAtIsNull().
			StatusNotIn(entity.LogBackupStatusStopped).
			PreloadBackupSchedule()
		return qs
	}
	logBackups, err := b.logBackupRepo.List(ctx, filterF)
	if err != nil {
		log.GetLogger(ctx).Error("list log backup error", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	if len(logBackups) == 0 {
		log.GetLogger(ctx).Error("no active log backup existed this should not happen", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	if len(logBackups) > 1 {
		log.GetLogger(ctx).Error("more than one active log backup existed this should not happen", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, cluster.GetClusterId())
	}
	return &logBackups[0], nil
}

func (b *pitBackupSwitcherImpl) Restart(ctx context.Context, req *domain.RestartPitBackupReq) (
	*domain.RestartPitBackupResp, bizerr.BizError,
) {
	logger := log.GetLogger(ctx).Named("pitBackupSwitcherImpl.Restart").With(
		zap.Any("cluster id", req.ClusterID))
	ctxWithLogger := log.WithLogger(ctx, logger)
	pitBackupSwitch, err := b.getOrCreatePitBackupSwitch(ctxWithLogger, req.ClusterID)
	if err != nil {
		return nil, err
	}
	nextStatus := b.statusMatrix.CalculateNextStatusForAction(status_matrix.CurrentStatus{
		Status:  string(pitBackupSwitch.Status),
		Actions: pitBackupSwitch.Action.CurrentActions,
	}, string(entity.PitBackupSwitchActionRestart))
	pitBackupSwitch.Action.CurrentActions = nextStatus.Actions
	pitBackupSwitch.Status = entity.PitBackupSwitchStatus(nextStatus.Status)
	if err := b.pitBackupSwitchRepo.Save(ctxWithLogger, pitBackupSwitch); err != nil {
		logger.Error("restart pit backup switch error", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	return &domain.RestartPitBackupResp{PitBackupSwitchID: pitBackupSwitch.ID.Uint64()}, nil
}

func (b *pitBackupSwitcherImpl) CancelRestart(
	ctx context.Context, req *domain.CancelRestartPitBackupReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx).
		Named("pitBackupSwitcherImpl.CancelRestart").
		With(zap.Any("cluster id", req.ClusterID))
	ctxWithLogger := log.WithLogger(ctx, logger)
	pitBackupSwitch, err := b.getOrCreatePitBackupSwitch(ctxWithLogger, req.ClusterID)
	if err != nil {
		return err
	}
	nextStatus := b.statusMatrix.CalculateNextStatusForAction(status_matrix.CurrentStatus{
		Status:  string(pitBackupSwitch.Status),
		Actions: pitBackupSwitch.Action.CurrentActions,
	}, string(entity.PitBackupSwitchActionCancelRestart))
	// filter restart action
	pitBackupSwitch.Action.CurrentActions = pie.Filter(pitBackupSwitch.Action.CurrentActions, func(action string) bool {
		return action != string(entity.PitBackupSwitchActionRestart)
	})
	pitBackupSwitch.Status = entity.PitBackupSwitchStatus(nextStatus.Status)
	if err := b.pitBackupSwitchRepo.Save(ctxWithLogger, pitBackupSwitch); err != nil {
		logger.Error("cancel restart pit backup switch error", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	return nil
}

func (b *pitBackupSwitcherImpl) GetOrCreatePitBackupSwitch(
	ctx context.Context, req *domain.SwitchPitBackupReq,
) (*domain.GetOrCreatePitBackupSwitchResp, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("pitBackupSwitcherImpl.Query").With(
		zap.Any("cluster id", req.ClusterID))
	ctxWithLogger := log.WithLogger(ctx, logger)
	pitBackupSwitch, err := b.getOrCreatePitBackupSwitch(ctxWithLogger, req.ClusterID)
	if err != nil {
		return nil, err
	}
	actions := b.statusMatrix.QueryAllowActions(status_matrix.CurrentStatus{
		Status:  string(pitBackupSwitch.Status),
		Actions: pitBackupSwitch.Action.CurrentActions,
	})
	return &domain.GetOrCreatePitBackupSwitchResp{
		PitBackupSwitch: pitBackupSwitch,
		AllowAction: pie.Map(actions, func(t string) entity.PitBackupSwitchActionType {
			return entity.PitBackupSwitchActionType(t)
		}),
	}, err
}