package auto

import (
	"github.com/samber/lo"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	crrfModel "github.com/tidbcloud/dataflow-service/app/models/crrf"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

// ── Enable from PiTR DAG ──────────────────────────────────────────────────

// AutoCRRFEnablePitParams are injected by the PiTR enable DAG via KeyPitBackupSwitchID.
type AutoCRRFEnablePitParams struct {
	PitBackupSwitchID uint64 `json:"pit_backup_switch_id"`
}

type autoCRRFEnableFromPitAction struct {
	autoCRRFEnableBase
	pitSwitchRepo backupRepo.PitBackupSwitchRepo
}

// NewAutoCRRFEnableFromPitAction creates the auto-enable action for insertion into the PiTR enable DAG.
func NewAutoCRRFEnableFromPitAction(
	backupSettingRepo backupRepo.BackupSettingRepo,
	pitSwitchRepo backupRepo.PitBackupSwitchRepo,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	crrfModel crrfModel.CRRFModel,
) *autoCRRFEnableFromPitAction {
	return &autoCRRFEnableFromPitAction{
		autoCRRFEnableBase: autoCRRFEnableBase{
			bsRepo:    backupSettingRepo,
			mgmtCli:   mgmtCli,
			crrfModel: crrfModel,
		},
		pitSwitchRepo: pitSwitchRepo,
	}
}

func (a *autoCRRFEnableFromPitAction) Name() string {
	return workflow.ActionAutoCRRFEnableFromPit
}

func (a *autoCRRFEnableFromPitAction) ParameterNew() interface{} {
	return &AutoCRRFEnablePitParams{}
}

func (a *autoCRRFEnableFromPitAction) Run(eCtx run.ExecuteContext, params interface{}) error {
	p := params.(*AutoCRRFEnablePitParams)
	pitSwitch, bizErr := a.pitSwitchRepo.Load(eCtx.Context(), p.PitBackupSwitchID)
	if bizErr != nil {
		return bizErr
	}
	return a.run(eCtx, uint64(pitSwitch.ClusterID))
}

// RunAfter polls QueryCRRF until the enable DAG reaches terminal state.
func (a *autoCRRFEnableFromPitAction) RunAfter(eCtx run.ExecuteContext, params interface{}) error {
	p := params.(*AutoCRRFEnablePitParams)
	pitSwitch, bizErr := a.pitSwitchRepo.Load(eCtx.Context(), p.PitBackupSwitchID)
	if bizErr != nil {
		return bizErr
	}
	return a.poll(eCtx, uint64(pitSwitch.ClusterID))
}

// ── Enable from dual-region DAG ───────────────────────────────────────────

// AutoCRRFEnableDualRegionParams are injected by the dual-region enable DAG via KeyBackupSettingId.
type AutoCRRFEnableDualRegionParams struct {
	BackupSettingID uint64 `json:"backup_setting_id"`
}

type autoCRRFEnableFromDualRegionAction struct {
	autoCRRFEnableBase
	pitSwitchRepo backupRepo.PitBackupSwitchRepo
}

// NewAutoCRRFEnableFromDualRegionAction creates the auto-enable action for dual-region related flows.
// Note:
//
//	For scenario 1(enable dual-region while PiTR is enabled): PiTR must be restarted first (workflow.DagRestartPitBackup), so CRRF auto-enable is triggered from that restart DAG to preserve ordering.
//	For scenario 2(enable-disable-enable dual-region while PiTR is enabled): PiTR restart is not needed because log backup is already running on dual-region bucket, so CRRF auto-enable is triggered from dual-region enable DAG directly.
func NewAutoCRRFEnableFromDualRegionAction(
	backupSettingRepo backupRepo.BackupSettingRepo,
	pitSwitchRepo backupRepo.PitBackupSwitchRepo,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	crrfModel crrfModel.CRRFModel,
) *autoCRRFEnableFromDualRegionAction {
	return &autoCRRFEnableFromDualRegionAction{
		autoCRRFEnableBase: autoCRRFEnableBase{
			bsRepo:    backupSettingRepo,
			mgmtCli:   mgmtCli,
			crrfModel: crrfModel,
		},
		pitSwitchRepo: pitSwitchRepo,
	}
}

func (a *autoCRRFEnableFromDualRegionAction) Name() string {
	return workflow.ActionAutoCRRFEnableFromDualRegion
}

func (a *autoCRRFEnableFromDualRegionAction) ParameterNew() interface{} {
	return &AutoCRRFEnableDualRegionParams{}
}

func (a *autoCRRFEnableFromDualRegionAction) Run(eCtx run.ExecuteContext, params interface{}) error {
	logger := action.GetLogger(eCtx)
	p := params.(*AutoCRRFEnableDualRegionParams)
	bs, bizErr := a.bsRepo.Load(eCtx.Context(), entity.ID(p.BackupSettingID))
	if bizErr != nil {
		return bizErr
	}

	pitSwitch, bizErr := a.pitSwitchRepo.LoadByClusterID(eCtx.Context(), uint64(bs.ClusterID))
	if bizErr != nil && bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return bizErr
	}
	// when pit backup switch is being restarted, the underlying logbackup is running on old storage bucket (single-region bucket) so that
	// the crrf daemon may fail to enable or run on old bucket. To address this case, we skip auto enabling CRRF in dual-region enable DAG and
	// let the pit backup restart DAG to trigger auto CRRF enable after pit backup is restarted and logbackup is running on new bucket.
	if pitSwitch != nil && lo.Contains([]entity.PitBackupSwitchStatus{entity.PitBackupStatusWaitingForRestarting, entity.PitBackupStatusRestarting}, pitSwitch.Status) {
		logger.Info("skip auto-enable CRRF in dual-region flow; PiTR is restarting and will trigger CRRF enable in PiTR restart DAG",
			zap.String("status", string(pitSwitch.Status)),
			zap.Uint64("clusterID", uint64(bs.ClusterID)),
		)
		return nil
	}

	return a.run(eCtx, uint64(bs.ClusterID))
}

// RunAfter polls QueryCRRF until the enable DAG reaches terminal state.
func (a *autoCRRFEnableFromDualRegionAction) RunAfter(eCtx run.ExecuteContext, params interface{}) error {
	p := params.(*AutoCRRFEnableDualRegionParams)
	bs, bizErr := a.bsRepo.Load(eCtx.Context(), entity.ID(p.BackupSettingID))
	if bizErr != nil {
		return bizErr
	}
	return a.poll(eCtx, uint64(bs.ClusterID))
}