package enable

import (
	"context"
	"fmt"

	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 enableDualRegionParams struct {
	ClusterID                uint64 `json:"cluster_id"`
	SecondaryRegion          string `json:"secondary_region"`
	SkipEnablePiTRDualRegion bool   `json:"skip_enable_pitr_dual_region,string"`
}

type enableDualRegionAction struct {
	backupSettingRepo  backupRepo.BackupSettingRepo
	backupSettingModel backupModel.BackupSettingModel
	crrfSettingRepo    crrfRepo.CRRFSettingRepo
}

// NewEnableDualRegionAction returns a Wrapper that triggers S3 CRR dual-region setup in Run
// and polls until the setup is active in RunAfter, then caches bucket IDs into CRRFSetting.
func NewEnableDualRegionAction(
	backupSettingRepo backupRepo.BackupSettingRepo,
	backupSettingModel backupModel.BackupSettingModel,
	crrfSettingRepo crrfRepo.CRRFSettingRepo,
) *action.Wrapper {
	return action.MustNewActionWrapper(
		workflow.ActionCRRFEnableDualRegion,
		&enableDualRegionAction{
			backupSettingRepo:  backupSettingRepo,
			backupSettingModel: backupSettingModel,
			crrfSettingRepo:    crrfSettingRepo,
		},
	)
}

func (a *enableDualRegionAction) Params() interface{} {
	return &enableDualRegionParams{}
}

// Run triggers S3 CRR dual-region setup. It is idempotent: if dual-region is already configured
// it 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 *enableDualRegionAction) Run(ctx context.Context, logger *zap.Logger, params interface{}) error {
	p := params.(*enableDualRegionParams)

	if p.SkipEnablePiTRDualRegion {
		logger.Info("skip_enable_pitr_dual_region=true: skipping dual-region 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 configured — CRR is active.
	if backupSetting.Bucket != nil && backupSetting.Bucket.DualRegionEnable {
		logger.Info("dual-region already configured, skipping trigger", zap.Uint64("clusterID", p.ClusterID))
		return nil
	}

	masterRegion, ok := backupSetting.MasterRegion()
	if !ok {
		logger.Error("master region not found in backup setting", zap.Uint64("clusterID", p.ClusterID))
		return action.StopRetry
	}

	if _, bizErr = a.backupSettingModel.ModifyRegionSetting(ctx, &domain.RegionSettingModificationReqV2{
		ClusterID: &p.ClusterID,
		Enable:    true,
		Regions:   []string{masterRegion, p.SecondaryRegion},
	}); bizErr != nil {
		logger.Warn("failed to trigger dual-region enable", zap.Error(bizErr))
		return bizErr
	}

	logger.Info("dual-region enable triggered",
		zap.Uint64("clusterID", p.ClusterID),
		zap.String("masterRegion", masterRegion),
		zap.String("secondaryRegion", p.SecondaryRegion))
	return nil
}

// RunAfter polls until dual-region is active and the backup setting is in normal status,
// then caches master and secondary bucket IDs into CRRFSetting.
// When SkipEnablePiTRDualRegion is true the step is skipped; EnableCRRF already populated
// the bucket IDs from the backup setting at the time the CRRF enable was requested.
// The Wrapper calls this with infinite retry until it succeeds or the task times out.
func (a *enableDualRegionAction) RunAfter(ctx context.Context, logger *zap.Logger, params interface{}) error {
	p := params.(*enableDualRegionParams)

	if p.SkipEnablePiTRDualRegion {
		logger.Info("skip_enable_pitr_dual_region=true: skipping dual-region RunAfter poll", 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
	}

	if backupSetting.Bucket == nil || !backupSetting.Bucket.DualRegionEnable {
		return fmt.Errorf("dual-region not yet enabled for cluster %d", p.ClusterID)
	}
	if backupSetting.Status != entity.BackupSettingStatusNormal {
		return fmt.Errorf("backup setting status is %s, waiting for normal", backupSetting.Status)
	}

	crrfSetting, bizErr := a.crrfSettingRepo.GetByClusterID(ctx, p.ClusterID)
	if bizErr != nil {
		return bizErr
	}

	for _, b := range backupSetting.Bucket.Buckets {
		if b.Master {
			crrfSetting.MasterBucketID = entity.ID(b.BucketId)
		} else {
			crrfSetting.SecondaryBucketID = entity.ID(b.BucketId)
		}
	}
	if saveErr := a.crrfSettingRepo.Save(ctx, crrfSetting); saveErr != nil {
		return saveErr
	}

	logger.Info("dual-region ready, bucket IDs cached",
		zap.Uint64("clusterID", p.ClusterID),
		zap.Uint64("masterBucketID", uint64(crrfSetting.MasterBucketID)),
		zap.Uint64("secondaryBucketID", uint64(crrfSetting.SecondaryBucketID)))
	return nil
}