package enable

import (
	"context"
	"fmt"
	"time"

	backupModel "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"
	crrfRepo "github.com/tidbcloud/dataflow-service/app/repo/crrf"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	"go.uber.org/zap"
)

type enablePitrParams struct {
	ClusterID                uint64 `json:"cluster_id"`
	SkipEnablePiTRDualRegion bool   `json:"skip_enable_pitr_dual_region,string"`
}

type enablePitrAction struct {
	backupSettingRepo  backupRepo.BackupSettingRepo
	backupSettingModel backupModel.BackupSettingModel
	logBackupRepo      backupRepo.LogBackupRepo
	crrfSettingRepo    crrfRepo.CRRFSettingRepo
}

// NewEnablePitrAction returns a Wrapper that triggers PiTR log backup in Run
// and polls until a running log backup is found in RunAfter.
func NewEnablePitrAction(
	backupSettingRepo backupRepo.BackupSettingRepo,
	backupSettingModel backupModel.BackupSettingModel,
	logBackupRepo backupRepo.LogBackupRepo,
	crrfSettingRepo crrfRepo.CRRFSettingRepo,
) *action.Wrapper {
	return action.MustNewActionWrapper(
		workflow.ActionCRRFEnablePitr,
		&enablePitrAction{
			backupSettingRepo:  backupSettingRepo,
			backupSettingModel: backupSettingModel,
			logBackupRepo:      logBackupRepo,
			crrfSettingRepo:    crrfSettingRepo,
		},
		action.SetRunAfterRetryFun(action.InvokeRunAfterWithExponentialRetryFactory(15*time.Second, 5*time.Minute, 0)),
	)
}

func (a *enablePitrAction) Params() interface{} {
	return &enablePitrParams{}
}

// Run triggers PiTR log backup. It is idempotent: if PiTR is already enabled, exits immediately.
// When SkipEnablePiTRDualRegion is true (called from inside a PiTR/dual-region enable DAG),
// the trigger step is skipped entirely to prevent a deadlock with the parent DAG.
// The Wrapper calls this with finite exponential retry.
func (a *enablePitrAction) Run(ctx context.Context, logger *zap.Logger, params interface{}) error {
	p := params.(*enablePitrParams)

	if p.SkipEnablePiTRDualRegion {
		logger.Info("skip_enable_pitr_dual_region=true: skipping PiTR trigger", zap.Uint64("clusterID", p.ClusterID))
		return nil
	}

	backupSetting, bizErr := a.backupSettingRepo.GetByFilter(ctx, &backupRepo.BackupSettingFilter{
		ClusterID: &p.ClusterID,
	})
	if bizErr != nil {
		return bizErr
	}
	if backupSetting == nil {
		logger.Error("backup setting not found", zap.Uint64("clusterID", p.ClusterID))
		return action.StopRetry
	}

	// Already enabled — nothing to trigger.
	if backupSetting.PiTR != nil && backupSetting.PiTR.Enable {
		logger.Info("PiTR already enabled, skipping trigger", zap.Uint64("clusterID", p.ClusterID))
		return nil
	}

	if bizErr = a.backupSettingModel.ModifyPitBackupSetting(ctx, &domain.PitSettingModificationReq{
		ClusterID: p.ClusterID,
		Enable:    true,
	}); bizErr != nil {
		logger.Warn("failed to trigger PiTR enable", zap.Error(bizErr))
		return bizErr
	}

	logger.Info("PiTR enable triggered", zap.Uint64("clusterID", p.ClusterID))
	return nil
}

// RunAfter polls until a running log backup is found, then records its ID in CRRFSetting.
// When SkipEnablePiTRDualRegion is true the entire step is skipped; LogBackupID will be
// resolved at runtime by LoadCRRFActionContext's fallback.
// The Wrapper calls this with infinite retry until it succeeds or the task times out.
func (a *enablePitrAction) RunAfter(ctx context.Context, logger *zap.Logger, params interface{}) error {
	p := params.(*enablePitrParams)

	if p.SkipEnablePiTRDualRegion {
		logger.Info("skip_enable_pitr_dual_region=true: skipping PiTR RunAfter poll", zap.Uint64("clusterID", p.ClusterID))
		return nil
	}

	logBackups, bizErr := a.logBackupRepo.List(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		return qs.ClusterIDEq(entity.ID(p.ClusterID)).StatusEq(entity.LogBackupStatusRunning).OrderDescByID()
	})
	if bizErr != nil {
		return bizErr
	}
	if len(logBackups) == 0 {
		return fmt.Errorf("PiTR log backup not yet running for cluster %d", p.ClusterID)
	}

	lb := logBackups[0]
	logger.Info("PiTR ready: running log backup found",
		zap.Uint64("clusterID", p.ClusterID),
		zap.Uint64("logBackupID", uint64(lb.ID)))

	setting, bizErr := a.crrfSettingRepo.GetByClusterID(ctx, p.ClusterID)
	if bizErr != nil {
		return bizErr
	}
	if setting.LogBackupID != lb.ID {
		setting.LogBackupID = lb.ID
		if saveErr := a.crrfSettingRepo.Save(ctx, setting); saveErr != nil {
			return saveErr
		}
		logger.Info("CRRFSetting.LogBackupID recorded",
			zap.Uint64("clusterID", p.ClusterID),
			zap.Uint64("logBackupID", uint64(lb.ID)))
	}
	return nil
}